Apache Airflow: ETL을 더 쉽게 만들기

안녕하세요, 저는 Vezet 그룹 회사 분석 부서의 데이터 엔지니어인 Dmitry Logvinenko입니다.

ETL 프로세스 개발을 위한 훌륭한 도구인 Apache Airflow에 대해 말씀드리겠습니다. 그러나 Airflow는 매우 다재다능하고 다면적이므로 데이터 흐름에 관여하지 않더라도 주기적으로 프로세스를 시작하고 실행을 모니터링해야 하는 경우에도 자세히 살펴봐야 합니다.

그리고 예, 말할뿐만 아니라 보여줄 것입니다. 프로그램에는 많은 코드, 스크린 샷 및 권장 사항이 있습니다.

Google에서 Airflow라는 단어를 검색하면 일반적으로 표시되는 내용 / Wikimedia Commons

차례

• 소개
• 주요 부분, 실용적인 (그리고 약간의 이론적인)
  • 왜 우리(그리고 당신)는
  • 클러스터 조립
  • 기본 개념
  • 우리는 작업을 생성합니다
  • 꽃에 대해 조금
  • 우리는 저부하를 로드합니다
  • 연결, 후크 및 기타 변수
  • 사용자 지정 연산자 구문 분석
  • 뿌릴까?
  • 합산
• 파트 최종, 참조 및 정보
  • 우리가 당신을 위해 모은 갈퀴
  • 더 많은 자동화 도구
• 참조

소개

Apache Airflow는 Django와 같습니다.

• 파이썬으로 작성
• 훌륭한 관리자 패널이 있습니다.
• 무한 확장 가능

- 단지 더 낫고 완전히 다른 목적을 위해 만들어졌습니다.

• 무제한 시스템에서 작업 실행 및 모니터링(Celery/Kubernetes 및 양심이 허용하는 만큼)
• 매우 쉽게 작성하고 이해할 수 있는 Python 코드에서 동적 워크플로 생성
• 기성품 구성 요소와 집에서 만든 플러그인을 모두 사용하여 데이터베이스와 API를 서로 연결하는 기능(매우 간단함).

다음과 같이 Apache Airflow를 사용합니다.

• 우리는 DWH 및 ODS(Vertica 및 Clickhouse가 있음)의 다양한 소스(많은 SQL Server 및 PostgreSQL 인스턴스, 애플리케이션 메트릭이 있는 다양한 API, 심지어 1C)에서 데이터를 수집합니다.
• 얼마나 발전했는지 cron, ODS에서 데이터 통합 ​​프로세스를 시작하고 유지 관리도 모니터링합니다.

최근까지 우리의 요구 사항은 32코어와 50GB RAM이 있는 하나의 작은 서버로 처리되었습니다. Airflow에서는 다음과 같이 작동합니다.

• 더 200 개 (실제로 우리가 작업을 채우는 워크플로),
• 각각 평균적으로 70개의 작업,
• 이 선함은 시작됩니다 (또한 평균적으로) 한 시간에 한 번.

그리고 우리가 어떻게 확장했는지에 대해서는 아래에 쓰겠지만 이제 우리가 해결할 über-problem을 정의해 보겠습니다.

각각 50개의 데이터베이스가 있는 XNUMX개의 소스 SQL Server가 있습니다. 각각 하나의 프로젝트 인스턴스이며 구조가 동일합니다(거의 모든 곳, mua-ha-ha). 즉, 각각 Orders 테이블(다행히도 해당 테이블 이름은 모든 비즈니스에 적용할 수 있습니다). 우리는 서비스 필드(소스 서버, 소스 데이터베이스, ETL 작업 ID)를 추가하여 데이터를 가져오고 순진하게 Vertica에 넣습니다.

가자!

주요 부분, 실용적인 (그리고 약간의 이론적인)

왜 우리(그리고 당신)는

나무가 크고 내가 단순했을 때 SQL-schik, 한 러시아 소매점에서 우리는 사용할 수 있는 두 가지 도구를 사용하여 ETL 프로세스, 즉 데이터 흐름을 사기쳤습니다.

• 인포매티카 파워 센터 - 자체 하드웨어, 자체 버전 관리 기능을 갖춘 극도로 확산되고 매우 생산적인 시스템. 나는 그 능력의 1%를 금지하는 신을 사용했습니다. 왜? 글쎄요, 우선 380 년대 어딘가에있는이 인터페이스는 정신적으로 우리에게 압력을가합니다. 둘째, 이 장치는 매우 멋진 프로세스, 격렬한 구성 요소 재사용 및 기타 매우 중요한 엔터프라이즈 트릭을 위해 설계되었습니다. Airbus AXNUMX / year의 날개와 같이 비용이 든다는 사실에 대해서는 아무 말도하지 않겠습니다.

  주의, 스크린샷은 30세 미만의 사람들에게 약간의 상처를 줄 수 있습니다.

  

• SQL 서버 통합 서버 - 우리는 프로젝트 내 흐름에서 이 동지를 사용했습니다. 사실, 우리는 이미 SQL Server를 사용하고 있으며 ETL 도구를 사용하지 않는 것이 어떻게든 비합리적일 것입니다. 그것의 모든 것이 좋습니다 : 인터페이스가 아름답고 진행률 보고서 ... 하지만 이것이 우리가 소프트웨어 제품을 좋아하는 이유가 아닙니다. 그것을 버전 dtsx (저장할 때 노드가 섞인 XML) 할 수 있지만 요점이 무엇입니까? 한 서버에서 다른 서버로 수백 개의 테이블을 드래그하는 작업 패키지를 만드는 것은 어떻습니까? 예, 백, 마우스 버튼을 클릭하면 집게 손가락이 스무 조각에서 떨어집니다. 하지만 확실히 더 패셔너블해 보입니다.

  

우리는 확실히 탈출구를 찾았습니다. 케이스 짝수 거의 자체 작성 SSIS 패키지 생성기에 왔습니다 ...

... 그리고 새로운 직업이 저를 찾았습니다. 그리고 Apache Airflow가 저를 추월했습니다.

ETL 프로세스 설명이 간단한 Python 코드라는 것을 알았을 때 나는 기뻐서 춤추지 않았습니다. 이것이 데이터 스트림의 버전이 지정되고 구분되는 방식이며, 수백 개의 데이터베이스에서 단일 구조의 테이블을 하나의 대상으로 쏟아 붓는 것은 13 또는 XNUMX개의 XNUMX인치 화면에서 Python 코드의 문제가 되었습니다.

클러스터 조립

완전한 유치원을 준비하지 말고 여기에서 Airflow, 선택한 데이터베이스, Celery 및 도크에 설명 된 기타 사례 설치와 같이 완전히 명백한 것에 대해 이야기하지 마십시오.

즉시 실험을 시작할 수 있도록 스케치했습니다. docker-compose.yml 여기서:

• 실제로 올리자 기류: 스케줄러, 웹서버. 꽃은 셀러리 작업을 모니터링하기 위해 그곳에서 회전할 것입니다(이미 apache/airflow:1.10.10-python3.7, 그러나 우리는 상관하지 않습니다)
• PostgreSQL, Airflow는 서비스 정보(스케줄러 데이터, 실행 통계 등)를 작성하고 Celery는 완료된 작업을 표시합니다.
• Redis, Celery의 태스크 브로커 역할을 합니다.
• 셀러리 일꾼, 작업을 직접 실행합니다.
• 폴더로 ./dags dags에 대한 설명과 함께 파일을 추가합니다. 즉석에서 픽업되므로 재채기를 할 때마다 전체 스택을 저글링할 필요가 없습니다.

어떤 곳에서는 예제의 코드가 완전히 표시되지 않지만(텍스트를 어지럽히지 않도록) 어딘가에서 프로세스에서 수정됩니다. 전체 작업 코드 예제는 저장소에서 찾을 수 있습니다. https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

version: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

services:
  # Redis as a Celery broker
  broker:
    image: redis:6.0.5-alpine

  # DB for the Airflow metadata
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Main container with Airflow Webserver, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <<: *airflow-config

      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'
      AIRFLOW__SCHEDULER__MAX_THREADS: 8

      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10

    depends_on:
      - airflow-db
      - broker

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Airflow Webserver
      - 8080:8080

  # Celery worker, will be scaled using `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <<: *airflow-config

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

참고 사항 :

• 작곡의 조립에서 나는 잘 알려진 이미지에 크게 의존했습니다. 퍼클/도커-기류 - 꼭 확인하세요. 아마도 당신은 당신의 삶에서 다른 것이 필요하지 않을 것입니다.
• 모든 기류 설정은 airflow.cfg, 뿐만 아니라 (개발자 덕분에) 환경 변수를 통해 악의적으로 이용했습니다.
• 당연히 프로덕션 준비가 되어 있지 않습니다. 일부러 컨테이너에 하트비트를 넣지 않았고 보안에 신경 쓰지 않았습니다. 그러나 나는 실험자들에게 적합한 최소한의 작업을 수행했습니다.
• 참고 :
  • dag 폴더는 스케줄러와 작업자 모두에 액세스할 수 있어야 합니다.
  • 모든 타사 라이브러리에도 동일하게 적용됩니다. 스케줄러와 작업자가 있는 시스템에 모두 설치해야 합니다.

이제 간단합니다.

$ docker-compose up --scale worker=3

모든 것이 상승하면 웹 인터페이스를 볼 수 있습니다.

• 기류 : http://127.0.0.1:8080/admin/
• 꽃: http://127.0.0.1:5555/dashboard

기본 개념

이 모든 "dags"에서 아무것도 이해하지 못했다면 여기에 짧은 사전이 있습니다.

• 스케줄러 -Airflow에서 가장 중요한 삼촌, 로봇이 사람이 아닌 열심히 작동하도록 제어: 일정 모니터링, dags 업데이트, 작업 시작.

  일반적으로 이전 버전에서는 메모리에 문제가 있었고(기억 상실이 아니라 누수) 레거시 매개변수가 구성에 남아 있었습니다. run_duration - 재시작 간격. 그러나 이제 모든 것이 괜찮습니다.

• DAG (일명 "dag") - "방향성 비순환 그래프", 그러나 그러한 정의는 소수의 사람들에게만 알려줄 것입니다. 그러나 실제로는 서로 상호 작용하는 작업을 위한 컨테이너(아래 참조)이거나 SSIS의 패키지 및 Informatica의 워크플로와 유사합니다. .

  dags 외에도 여전히 하위 dag가 있을 수 있지만 대부분 도달하지 못할 것입니다.

• DAG 실행 - 자체적으로 할당된 초기화된 dag execution_date. 같은 dag의 Dagrans는 병렬로 작업할 수 있습니다(물론 작업을 멱등성으로 만든 경우).
• 운영자 특정 작업을 수행하는 코드 조각입니다. 세 가지 유형의 연산자가 있습니다.
  • 동작우리가 가장 좋아하는 것처럼 PythonOperator모든 (유효한) Python 코드를 실행할 수 있습니다.
  • 이전, 예를 들어 장소에서 장소로 데이터를 전송하는 MsSqlToHiveTransfer;
  • 감지기 반면에 이벤트가 발생할 때까지 dag의 추가 실행에 반응하거나 속도를 늦출 수 있습니다. HttpSensor 지정된 끝점을 가져올 수 있으며 원하는 응답이 대기 중일 때 전송을 시작합니다. GoogleCloudStorageToS3Operator. 호기심 많은 마음은 "왜? 결국 연산자에서 바로 반복을 수행할 수 있습니다!” 그런 다음 정지된 운영자로 작업 풀을 막히지 않도록 합니다. 센서는 다음 시도 전에 시작, 확인 및 종료됩니다.
• 태스크 - 선언된 연산자는 유형에 관계없이 dag에 부착되어 작업의 순위로 승격됩니다.
• 작업 인스턴스 -일반 기획자가 수행자-작업자에 대한 전투에 작업을 보낼 시간이라고 결정했을 때 (우리가 사용하는 경우 바로 그 자리에서) LocalExecutor 또는 원격 노드의 경우 CeleryExecutor) 컨텍스트를 할당하고(즉, 변수 세트 - 실행 매개변수) 명령 또는 쿼리 템플릿을 확장하고 풀링합니다.

우리는 작업을 생성합니다

먼저 doug의 일반적인 계획을 개략적으로 설명한 다음 사소한 솔루션을 적용하기 때문에 세부 사항을 점점 더 자세히 살펴 보겠습니다.

따라서 가장 간단한 형태로 이러한 dag는 다음과 같습니다.

from datetime import timedelta, datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

from commons.datasources import sql_server_ds

dag = DAG('orders',
          schedule_interval=timedelta(hours=6),
          start_date=datetime(2020, 7, 8, 0))

def workflow(**context):
    print(context)

for conn_id, schema in sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

그것을 알아 봅시다 :

• 먼저 필요한 라이브러리를 가져오고 다른 것;
• sql_server_ds -가요 List[namedtuple[str, str]] Airflow Connections의 연결 이름과 플레이트를 가져올 데이터베이스;
• dag -반드시 있어야하는 우리 dag의 발표 globals(), 그렇지 않으면 Airflow에서 찾을 수 없습니다. Doug는 또한 다음과 같이 말할 필요가 있습니다.
  • 그의 이름은 무엇입니까 orders - 그러면 이 이름이 웹 인터페이스에 표시됩니다.
  • 그는 XNUMX월 XNUMX일 자정부터 일할 것이며,
  • 약 6시간마다 실행되어야 합니다. timedelta() 허용 cron-선 0 0 0/6 ? * * *, 덜 멋진 경우 - 다음과 같은 표현 @daily);
• workflow() 주요 작업을 수행하지만 지금은 아닙니다. 지금은 컨텍스트를 로그에 덤프합니다.
• 이제 작업 생성의 간단한 마법:
  • 우리는 소스를 통해 실행합니다.
  • 초기화 PythonOperator, 더미를 실행할 것입니다. workflow(). 작업의 고유한(dag 내) 이름을 지정하고 dag 자체를 연결하는 것을 잊지 마십시오. 깃발 provide_context 차례로 함수에 추가 인수를 추가하고 다음을 사용하여 신중하게 수집합니다. **context.

지금은 그게 다입니다. 우리가 얻은 것:

• 웹 인터페이스의 새로운 dag,
• 병렬로 실행될 XNUMX개의 작업(Airflow, Celery 설정 및 서버 용량이 허용하는 경우).

글쎄, 거의 알았어요.

누가 종속성을 설치합니까?

이 모든 것을 단순화하기 위해 나는 망쳤습니다 docker-compose.yml 가공 requirements.txt 모든 노드에서.

이제 사라졌습니다.

회색 사각형은 스케줄러가 처리하는 작업 인스턴스입니다.

우리는 조금 기다리면 작업이 작업자에 의해 스냅됩니다.

물론 녹색은 성공적으로 작업을 완료했습니다. Reds는 그다지 성공적이지 않습니다.

그런데 우리 제품에는 폴더가 없습니다. ./dags, 기계간에 동기화가 없습니다. 모든 dag가 있습니다. git Gitlab에서 Gitlab CI는 병합할 때 컴퓨터에 업데이트를 배포합니다. master.

꽃에 대해 조금

일꾼들이 젖꼭지를 때리는 동안 우리에게 무엇인가를 보여줄 수 있는 또 다른 도구인 꽃을 기억해 봅시다.

작업자 노드에 대한 요약 정보가 있는 첫 번째 페이지:

작업이 진행된 가장 집중적인 페이지:

브로커 상태가 있는 가장 지루한 페이지:

가장 밝은 페이지는 작업 상태 그래프와 실행 시간이 있는 페이지입니다.

우리는 저부하를 로드합니다

따라서 모든 작업이 완료되었으므로 부상자를 옮길 수 있습니다.

그리고 여러 가지 이유로 부상자가 많았습니다. Airflow를 올바르게 사용한 경우 바로 이러한 사각형은 데이터가 확실히 도착하지 않았음을 나타냅니다.

로그를 보고 타락한 작업 인스턴스를 다시 시작해야 합니다.

사각형을 클릭하면 사용 가능한 작업이 표시됩니다.

쓰러진 자들을 데려가서 클리어할 수 있습니다. 즉, 거기에서 무언가 실패했다는 사실을 잊고 동일한 인스턴스 작업이 스케줄러로 이동합니다.

모든 빨간색 사각형이 있는 마우스로 이 작업을 수행하는 것은 그다지 인도적이지 않다는 것이 분명합니다. 이것은 Airflow에서 기대하는 것이 아닙니다. 당연히 우리에게는 대량 살상 무기가 있습니다. Browse/Task Instances

한 번에 모든 항목을 선택하고 XNUMX으로 재설정하고 올바른 항목을 클릭합니다.

청소 후 택시는 다음과 같습니다(스케줄러가 예약하기를 이미 기다리고 있음).

연결, 후크 및 기타 변수

다음 DAG를 살펴볼 시간입니다. update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='DWH Reports updated',
    html_content=dedent("""Господа хорошие, отчеты обновлены"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Наташ, просыпайся, мы {{ dag.dag_id }} уронили
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

모두가 보고서 업데이트를 수행한 적이 있습니까? 이것은 다시 그녀입니다. 데이터를 가져올 소스 목록이 있습니다. 넣을 목록이 있습니다. 모든 일이 발생하거나 중단되었을 때 경적을 울리는 것을 잊지 마십시오. (음, 이것은 우리에 관한 것이 아닙니다.)

파일을 다시 살펴보고 모호한 새 항목을 살펴보겠습니다.

• from commons.operators import TelegramBotSendMessage - 우리가 Unblocked에 메시지를 보내기 위한 작은 래퍼를 만들어 활용한 자체 연산자를 만드는 것을 방해하는 것은 없습니다. (아래에서 이 연산자에 대해 자세히 설명하겠습니다.)
• default_args={} - dag는 모든 연산자에 동일한 인수를 배포할 수 있습니다.
• to='{{ var.value.all_the_kings_men }}' - 필드 to 우리는 하드코딩하지 않고 Jinja와 내가 조심스럽게 입력한 이메일 목록이 있는 변수를 사용하여 동적으로 생성합니다. Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS - 오퍼레이터 시작 조건. 우리의 경우 모든 종속성이 해결된 경우에만 편지가 상사에게 전달됩니다. 성공적으로;
• tg_bot_conn_id='tg_main' - 인수 conn_id 우리가 만든 연결 ID를 수락합니다. Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED - Telegram의 메시지는 떨어진 작업이 있는 경우에만 날아갑니다.
• task_concurrency=1 - 하나의 작업에 대한 여러 작업 인스턴스의 동시 시작을 금지합니다. 그렇지 않으면 여러 제품을 동시에 출시할 것입니다. VerticaOperator (하나의 테이블을 보고);
• report_update >> [email, tg] - 모두 VerticaOperator 다음과 같이 편지와 메시지를 보내는 데 수렴합니다.
  

  그러나 알림 연산자는 시작 조건이 다르기 때문에 하나만 작동합니다. 트리 보기에서는 모든 것이 덜 시각적으로 보입니다.
  

나는에 대해 몇 마디 말할 것이다 매크로 그리고 그들의 친구들 - 변수.

매크로는 다양한 유용한 정보를 연산자 인수로 대체할 수 있는 Jinja 자리 표시자입니다. 예를 들면 다음과 같습니다.

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} 컨텍스트 변수의 내용으로 확장됩니다. execution_date 형식으로 YYYY-MM-DD: 2020-07-14. 가장 좋은 점은 컨텍스트 변수가 특정 작업 인스턴스(트리 보기의 사각형)에 고정되고 다시 시작하면 자리 표시자가 동일한 값으로 확장된다는 것입니다.

할당된 값은 각 작업 인스턴스의 렌더링됨 버튼을 사용하여 볼 수 있습니다. 편지를 보내는 작업은 다음과 같습니다.

따라서 메시지를 보내는 작업에서 다음을 수행합니다.

사용 가능한 최신 버전의 내장 매크로 전체 목록은 여기에서 확인할 수 있습니다. 매크로 참조

또한 플러그인의 도움으로 우리는 우리 자신의 매크로를 선언할 수 있지만 그것은 또 다른 이야기입니다.

미리 정의된 것 외에도 변수의 값을 대체할 수 있습니다(이미 위의 코드에서 사용했습니다). 에서 생성하자 Admin/Variables 몇 가지:

사용할 수 있는 모든 것:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

값은 스칼라이거나 JSON일 수도 있습니다. JSON의 경우:

bot_config

{
    "bot": {
        "token": 881hskdfASDA16641,
        "name": "Verter"
    },
    "service": "TG"
}

원하는 키의 경로를 사용하십시오. {{ var.json.bot_config.bot.token }}.

나는 문자 그대로 한 단어를 말하고 하나의 스크린 샷을 보여줍니다. 연결. 여기에서는 모든 것이 기본입니다. 페이지에서 Admin/Connections 우리는 연결을 생성하고 거기에 로그인/암호 및 보다 구체적인 매개 변수를 추가합니다. 이와 같이:

암호를 암호화하거나(기본값보다 더 철저하게) 연결 유형을 생략할 수 있습니다(예: tg_main) - 사실은 유형 목록이 Airflow 모델에 고정되어 있고 소스 코드에 들어가지 않고는 확장할 수 없다는 것입니다(갑자기 Google에 무언가를 검색하지 않은 경우 수정해 주세요). 이름.

동일한 이름으로 여러 연결을 만들 수도 있습니다. 이 경우 메서드는 BaseHook.get_connection(), 이름으로 우리를 연결시켜 줄 것입니다. 무작위의 여러 이름에서 따왔습니다(Round Robin을 만드는 것이 더 논리적이지만 Airflow 개발자의 양심에 맡기겠습니다).

변수 및 연결은 확실히 멋진 도구이지만 균형을 잃지 않는 것이 중요합니다. 코드 자체에 저장하는 흐름 부분과 저장을 위해 Airflow에 제공하는 부분. 한편으로는 UI를 통해 우편함과 같은 값을 빠르게 변경하는 것이 편리할 수 있습니다. 반면에 이것은 여전히 ​​우리가 제거하고 싶었던 마우스 클릭으로의 복귀입니다.

연결 작업은 작업 중 하나입니다. 후크. 일반적으로 Airflow hook은 이를 타사 서비스 및 라이브러리에 연결하기 위한 지점입니다. 예를 들어, JiraHook Jira와 상호 작용할 수 있도록 클라이언트를 열 것이며(작업을 앞뒤로 이동할 수 있음) SambaHook 로컬 파일을 smb-가리키다.

사용자 지정 연산자 구문 분석

그리고 우리는 그것이 어떻게 만들어지는지 거의 볼 수 있었습니다. TelegramBotSendMessage

암호 commons/operators.py 실제 연산자:

from typing import Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Send message to chat_id using TelegramBotHook

    Example:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Union[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Send "{self.message}" to the chat {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

여기에서 Airflow의 다른 모든 것과 마찬가지로 모든 것이 매우 간단합니다.

• 에서 상속 BaseOperator, 꽤 많은 Airflow 관련 사항을 구현합니다(여가를 보십시오).
• 선언된 필드 template_fields, 여기서 Jinja는 처리할 매크로를 찾습니다.
• 에 대한 올바른 주장을 정리했습니다. __init__(), 필요한 경우 기본값을 설정합니다.
• 조상의 초기화도 잊지 않았습니다.
• 해당 후크를 열었습니다. TelegramBotHook그것으로부터 클라이언트 개체를 받았습니다.
• 재정의(재정의) 메서드 BaseOperator.execute(), Airfow는 운영자를 시작할 때가되면 트 위치합니다. 로그인하는 것을 잊고 기본 작업을 구현합니다. (그런데, 바로 로그인합니다. stdout и stderr - 기류가 모든 것을 가로막고 아름답게 감싸고 필요한 곳에 분해합니다.)

우리가 가진 것을 보자 commons/hooks.py. 후크 자체가 포함된 파일의 첫 번째 부분:

from typing import Union

from airflow.hooks.base_hook import BaseHook
from requests_toolbelt.sessions import BaseUrlSession

class TelegramBotHook(BaseHook):
    """Telegram Bot API hook

    Note: add a connection with empty Conn Type and don't forget
    to fill Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

여기에서 무엇을 설명해야할지조차 모르겠고 중요한 사항만 언급하겠습니다.

• 우리는 상속하고 인수에 대해 생각합니다. 대부분의 경우 다음 중 하나입니다. conn_id;
• 표준 방법 재정의: 나는 스스로를 제한했다 get_conn(), 이름으로 연결 매개변수를 가져오고 섹션만 가져옵니다. extra (이것은 JSON 필드입니다) (내 자신의 지침에 따라!) Telegram 봇 토큰을 넣습니다. {"bot_token": "YOuRAwEsomeBOtToKen"}.
• 나는 우리의 인스턴스를 생성 TelegramBot, 특정 토큰을 제공합니다.

그게 다야. 다음을 사용하여 후크에서 클라이언트를 가져올 수 있습니다. TelegramBotHook().clent 또는 TelegramBotHook().get_conn().

그리고 Telegram REST API에 대한 마이크로 랩퍼를 만드는 파일의 두 번째 부분은 동일한 항목을 드래그하지 않도록 합니다. python-telegram-bot 한 가지 방법 sendMessage.

class TelegramBot:
    """Telegram Bot API wrapper

    Examples:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hi, darling')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hi, darling', chat_id=-1762374628374)
    """
    API_ENDPOINT = 'https://api.telegram.org/bot{}/'

    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):
        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)
        self.session = BaseUrlSession(self._base_url)
        self.chat_id = chat_id

    def send_message(self, message: str, chat_id: Union[int, str] = None):
        method = 'sendMessage'

        payload = {'chat_id': chat_id or self.chat_id,
                   'text': message,
                   'parse_mode': 'MarkdownV2'}

        response = self.session.post(method, data=payload).json()
        if not response.get('ok'):
            raise TelegramBotException(response)

class TelegramBotException(Exception):
    def __init__(self, *args, **kwargs):
        super().__init__((args, kwargs))

올바른 방법은 모두 더하는 것입니다. TelegramBotSendMessage, TelegramBotHook, TelegramBot - 플러그인에서 공개 저장소에 넣고 오픈 소스에 제공하십시오.

이 모든 것을 연구하는 동안 보고서 업데이트가 성공적으로 실패하고 채널에서 오류 메시지를 보냈습니다. 잘못된건 아닌지 확인해봐야겠네요...

우리 개에서 뭔가 고장났습니다! 그것이 우리가 기대했던 것이 아닙니까? 정확히!

뿌릴까?

내가 뭔가를 놓친 것 같니? 그는 SQL Server에서 Vertica로 데이터를 전송하겠다고 약속한 다음 데이터를 가져와 악당이라는 주제에서 벗어났습니다!

이 잔학 행위는 의도적이었습니다. 나는 당신을 위해 몇 가지 용어를 해독해야 했습니다. 이제 더 멀리 갈 수 있습니다.

우리의 계획은 이랬습니다.

1. 대그를 하다
2. 작업 생성
3. 모든 것이 얼마나 아름다운지보십시오
4. 채우기에 세션 번호 할당
5. SQL Server에서 데이터 가져오기
6. Vertica에 데이터 넣기
7. 통계 수집

그래서 이 모든 것을 시작하고 실행하기 위해 저는 우리 docker-compose.yml:

docker-compose.db.yml

version: '3.4'

x-mssql-base: &mssql-base
  image: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04
  restart: always
  environment:
    ACCEPT_EULA: Y
    MSSQL_PID: Express
    SA_PASSWORD: SayThanksToSatiaAt2020
    MSSQL_MEMORY_LIMIT_MB: 1024

services:
  dwh:
    image: jbfavre/vertica:9.2.0-7_ubuntu-16.04

  mssql_0:
    <<: *mssql-base

  mssql_1:
    <<: *mssql-base

  mssql_2:
    <<: *mssql-base

  mssql_init:
    image: mio101/py3-sql-db-client-base
    command: python3 ./mssql_init.py
    depends_on:
      - mssql_0
      - mssql_1
      - mssql_2
    environment:
      SA_PASSWORD: SayThanksToSatiaAt2020
    volumes:
      - ./mssql_init.py:/mssql_init.py
      - ./dags/commons/datasources.py:/commons/datasources.py

거기에서 우리는 다음을 제기합니다.

• Vertica를 호스트로 dwh 가장 기본적인 설정으로
• 세 개의 SQL Server 인스턴스,
• 우리는 후자의 데이터베이스를 일부 데이터로 채웁니다(어떠한 경우에도 mssql_init.py!)

우리는 지난 번보다 약간 더 복잡한 명령의 도움으로 모든 좋은 것을 시작합니다.

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

우리의 Miracle Randomizer가 생성한 항목을 사용할 수 있습니다. Data Profiling/Ad Hoc Query:

가장 중요한 것은 분석가에게 보여주지 않는 것입니다.

~에 대해 자세하게 말하다 ETL 세션 거기에는 모든 것이 사소합니다. 베이스를 만들고 그 안에 사인이 있고 컨텍스트 관리자로 모든 것을 래핑하고 이제 다음을 수행합니다.

with Session(task_name) as session:
    print('Load', session.id, 'started')

    # Load worflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

from sys import stderr

class Session:
    """ETL workflow session

    Example:
        with Session(task_name) as session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Well done'
    """

    def __init__(self, connection, task_name):
        self.connection = connection
        self.connection.autocommit = True

        self._task_name = task_name
        self._id = None

        self.loaded_rows = None
        self.successful = None
        self.comment = None

    def __enter__(self):
        return self.open()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if any(exc_type, exc_val, exc_tb):
            self.successful = False
            self.comment = f'{exc_type}: {exc_val}n{exc_tb}'
            print(exc_type, exc_val, exc_tb, file=stderr)
        self.close()

    def __repr__(self):
        return (f'<{self.__class__.__name__} ' 
                f'id={self.id} ' 
                f'task_name="{self.task_name}">')

    @property
    def task_name(self):
        return self._task_name

    @property
    def id(self):
        return self._id

    def _execute(self, query, *args):
        with self.connection.cursor() as cursor:
            cursor.execute(query, args)
            return cursor.fetchone()[0]

    def _create(self):
        query = """
            CREATE TABLE IF NOT EXISTS sessions (
                id          SERIAL       NOT NULL PRIMARY KEY,
                task_name   VARCHAR(200) NOT NULL,

                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,
                finished    TIMESTAMPTZ           DEFAULT current_timestamp,
                successful  BOOL,

                loaded_rows INT,
                comment     VARCHAR(500)
            );
            """
        self._execute(query)

    def open(self):
        query = """
            INSERT INTO sessions (task_name, finished)
            VALUES (%s, NULL)
            RETURNING id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'opened')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Session is not open')
        query = """
            UPDATE sessions
            SET
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            WHERE
                id = %s
            RETURNING id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'closed',
              ', successful: ', self.successful,
              ', Loaded: ', self.loaded_rows,
              ', comment:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

때가 왔다 데이터 수집 우리의 XNUMX개 테이블에서. 매우 소박한 라인의 도움으로 이것을 해보자:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()

query = f"""
    SELECT 
        id, start_time, end_time, type, data
    FROM dbo.Orders
    WHERE
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)

1. Airflow에서 얻은 후크의 도움으로 pymssql-연결하다
2. 날짜 형식의 제한을 요청으로 대체해 보겠습니다. 제한은 템플릿 엔진에 의해 함수에 던져집니다.
3. 우리의 요청을 공급 pandas누가 우리를 얻을 것인가 DataFrame - 앞으로 우리에게 유용할 것입니다.

나는 대체를 사용하고 있습니다 {dt} 요청 매개변수 대신 %s 내가 사악한 피노키오라서가 아니라 pandas 대처할 수 없다. pymssql 그리고 마지막을 슬쩍 params: List그가 정말로 원하지만 tuple.
개발자도 참고하세요 pymssql 더 이상 그를 지원하지 않기로 결정했고 이제 이사를 가야 할 때입니다 pyodbc.

Airflow가 함수의 인수를 무엇으로 채웠는지 살펴보겠습니다.

데이터가 없으면 계속할 필요가 없습니다. 그러나 채우기가 성공적이라고 생각하는 것도 이상합니다. 그러나 이것은 실수가 아닙니다. 아아아, 어떡해?! 다음은 다음과 같습니다.

if df.empty:
    raise AirflowSkipException('No rows to load')

AirflowSkipException Airflow에 오류가 없다고 알려주지만 작업을 건너뜁니다. 인터페이스에는 녹색 또는 빨간색 사각형이 아니라 분홍색이 있습니다.

데이터를 던지자 여러 열:

df['etl_source'] = src_schema
df['etl_id'] = session.id
df['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

즉

• 우리가 주문을 받은 데이터베이스,
• 플러딩 세션의 ID(다를 것입니다. 모든 작업에 대해),
• 소스 및 주문 ID의 해시 - 최종 데이터베이스(모든 것이 하나의 테이블에 쏟아지는 곳)에서 고유한 주문 ID를 갖게 됩니다.

마지막 두 번째 단계가 남아 있습니다. 모든 것을 Vertica에 붓습니다. 그리고 이상하게도 이를 수행하는 가장 훌륭하고 효율적인 방법 중 하나는 CSV를 사용하는 것입니다!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

1. 우리는 특별한 수신기를 만들고 있습니다 StringIO.
2. pandas 친절하게 우리의 DataFrame 의 형태로 CSV-윤곽.
3. 후크로 우리가 가장 좋아하는 Vertica에 대한 연결을 열어 봅시다.
4. 이제 도움을 받아 copy() 데이터를 Vertika로 직접 보내십시오!

우리는 드라이버에서 얼마나 많은 라인이 채워졌는지 가져오고 세션 관리자에게 모든 것이 정상이라고 알릴 것입니다.

session.loaded_rows = cursor.rowcount
session.successful = True

그게 다야.

판매시 타겟 플레이트를 수동으로 생성합니다. 여기에서 나는 작은 기계를 허용했습니다.

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

내가 사용하고 VerticaOperator() 데이터베이스 스키마와 테이블을 생성합니다(물론 아직 존재하지 않는 경우). 가장 중요한 것은 종속성을 올바르게 정렬하는 것입니다.

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

합산

-글쎄,-작은 쥐가 말했다.-그렇지, 지금
내가 숲에서 가장 끔찍한 동물이라고 확신합니까?

줄리아 도날드슨, Gruffalo

동료들과 경쟁이 있었다면 처음부터 ETL 프로세스를 신속하게 생성하고 시작할 사람 : 그들은 SSIS와 마우스를 사용하고 Airflow를 사용하면 ... 그리고 유지 관리의 용이성을 비교할 것입니다 ... 와우, 내가 모든 면에서 그들을 이길 것이라는 데 동의하실 것 같아요!

조금 더 진지하게 말하자면, 프로그램 코드의 형태로 프로세스를 설명함으로써 Apache Airflow가 제 일을 했습니다. 많이 더 편하고 즐겁게.

플러그인 및 확장성 측면에서 무제한 확장성은 거의 모든 영역에서 Airflow를 사용할 수 있는 기회를 제공합니다. 데이터 수집, 준비 및 처리의 전체 주기, 로켓 발사(화성, 강의).

파트 최종, 참조 및 정보

우리가 당신을 위해 모은 갈퀴

• start_date. 예, 이것은 이미 지역 밈입니다. Doug의 주요 주장을 통해 start_date 모두 통과 함. 간단히 말해서, 당신이 지정하는 경우 start_date 현재 날짜 및 schedule_interval - 언젠가는 DAG가 내일 더 일찍 시작되지 않습니다.

  start_date = datetime(2020, 7, 7, 0, 1, 2)

  그리고 더 이상 문제가 없습니다.

  이와 관련된 또 다른 런타임 오류가 있습니다. Task is missing the start_date parameter, 이것은 dag 연산자에 바인딩하는 것을 잊었다는 것을 가장 자주 나타냅니다.

• 모두 하나의 기계에 있습니다. 예, 베이스(Airflow 자체 및 코팅), 웹 서버, 스케줄러 및 작업자. 그리고 그것은 효과가 있었습니다. 그러나 시간이 지남에 따라 서비스에 대한 작업 수가 증가했고 PostgreSQL이 20ms가 아닌 5s에서 인덱스에 응답하기 시작했을 때 우리는 그것을 가져 와서 가져갔습니다.
• 로컬 실행자. 예, 우리는 여전히 그 위에 앉아 있고 이미 심연의 가장자리에 왔습니다. 지금까지는 LocalExecutor로도 충분했지만 이제 적어도 한 명의 작업자로 확장해야 할 때이며 CeleryExecutor로 이동하기 위해 열심히 노력해야 할 것입니다. 그리고 하나의 시스템에서 작업할 수 있다는 사실을 고려할 때 서버에서도 셀러리를 사용하는 것을 막을 수 있는 것은 없습니다.
• 미사용 내장 도구:
  • 연결 서비스 자격 증명을 저장하기 위해
  • SLA 누락 제시간에 해결되지 않은 작업에 응답하기 위해
  • 엑스컴 메타 데이터 교환을 위해 (내가 말했다 메타데이터!) dag 작업 사이.
• 메일 남용. 글쎄, 내가 뭐라고 말할 수 있니? 넘어진 작업의 모든 반복에 대해 경고가 설정되었습니다. 이제 내 작업 Gmail에는 Airflow에서 보낸 90개 이상의 이메일이 있고 웹 메일 총구는 한 번에 100개 이상을 선택하고 삭제하는 것을 거부합니다.

더 많은 함정: Apache Airflow 함정

더 많은 자동화 도구

우리가 손이 아닌 머리로 더 많이 일할 수 있도록 Airflow는 다음과 같이 준비했습니다.

• REST API - 그는 여전히 작업을 방해하지 않는 실험적 상태를 유지합니다. 이를 통해 dag 및 작업에 대한 정보를 얻을 수 있을 뿐만 아니라 dag를 중지/시작하고 DAG 실행 또는 풀을 만들 수 있습니다.
• CLI - WebUI를 통해 사용하기 불편할 뿐만 아니라 일반적으로 없는 많은 도구를 명령줄을 통해 사용할 수 있습니다. 예를 들어:
  • backfill 작업 인스턴스를 다시 시작하는 데 필요합니다.
    예를 들어, 분석가들이 와서 이렇게 말했습니다. 고쳐라, 고쳐라, 고쳐라, 고쳐라!" 그리고 당신은 그런 호브입니다.

    airflow backfill -s '2020-01-01' -e '2020-01-13' orders

  • 기본 서비스: initdb, resetdb, upgradedb, checkdb.
  • run, 하나의 인스턴스 작업을 실행하고 모든 종속성에 대해 점수를 매길 수 있습니다. 또한 다음을 통해 실행할 수 있습니다. LocalExecutor, 셀러리 클러스터가 있는 경우에도 마찬가지입니다.
  • 거의 같은 일을 한다 test, 기지에서만 아무것도 쓰지 않습니다.
  • connections 셸에서 연결을 대량으로 생성할 수 있습니다.
• 파이썬 API -플러그인을 위한 다소 하드코어한 상호 작용 방식이며 작은 손으로 무리를 짓지 않습니다. 하지만 누가 우리가 가는 걸 막겠어 /home/airflow/dags, 달리다 ipython 그리고 장난을 시작? 예를 들어 다음 코드를 사용하여 모든 연결을 내보낼 수 있습니다.

  from airflow import settings
  from airflow.models import Connection
  
  fields = 'conn_id conn_type host port schema login password extra'.split()
  
  session = settings.Session()
  for conn in session.query(Connection).order_by(Connection.conn_id):
    d = {field: getattr(conn, field) for field in fields}
    print(conn.conn_id, '=', d)

• Airflow 메타데이터베이스에 연결 중입니다. 여기에 작성하는 것은 권장하지 않지만 다양한 특정 메트릭에 대한 작업 상태를 가져오는 것이 API를 통하는 것보다 훨씬 빠르고 쉬울 수 있습니다.

  모든 작업이 idempotent는 아니지만 때때로 떨어질 수 있으며 이는 정상입니다. 그러나 몇 가지 막힘은 이미 의심스럽고 확인이 필요합니다.

  SQL에 주의하세요!

  WITH last_executions AS (
  SELECT
      task_id,
      dag_id,
      execution_date,
      state,
          row_number()
          OVER (
              PARTITION BY task_id, dag_id
              ORDER BY execution_date DESC) AS rn
  FROM public.task_instance
  WHERE
      execution_date > now() - INTERVAL '2' DAY
  ),
  failed AS (
      SELECT
          task_id,
          dag_id,
          execution_date,
          state,
          CASE WHEN rn = row_number() OVER (
              PARTITION BY task_id, dag_id
              ORDER BY execution_date DESC)
                   THEN TRUE END AS last_fail_seq
      FROM last_executions
      WHERE
          state IN ('failed', 'up_for_retry')
  )
  SELECT
      task_id,
      dag_id,
      count(last_fail_seq)                       AS unsuccessful,
      count(CASE WHEN last_fail_seq
          AND state = 'failed' THEN 1 END)       AS failed,
      count(CASE WHEN last_fail_seq
          AND state = 'up_for_retry' THEN 1 END) AS up_for_retry
  FROM failed
  GROUP BY
      task_id,
      dag_id
  HAVING
      count(last_fail_seq) > 0

참조

물론 Google 발행의 처음 XNUMX개 링크는 내 책갈피의 Airflow 폴더 내용입니다.

• Apache Airflow 설명서 - 물론 사무실부터 시작해야 합니다. 설명서는 있지만 지침을 읽는 사람은 누구입니까?
• 모범 사례 -글쎄요, 최소한 제작자의 권장 사항을 읽으십시오.
• 기류 UI - 맨 처음: 사진의 사용자 인터페이스
• Apache Airflow의 주요 개념 이해 -기본 개념이 잘 설명되어 있습니다. 만약 (갑자기!) 제 말을 이해하지 못하셨다면요.
• Tianlong의 블로그 — Airflow 서버/클러스터 구축 방법에 대한 가이드 - Airflow 클러스터 설정을 위한 간단한 가이드입니다.
• Lyft에서 Apache Airflow 실행 - 아마도 더 많은 형식주의와 더 적은 예를 제외하고는 거의 동일한 흥미로운 기사입니다.
• Apache Airflow가 Celery 작업자에 작업을 배포하는 방법 — 셀러리와 함께 일하는 것에 대해.
• Apache Airflow의 DAG 작성 모범 사례 - 작업의 멱등성, 날짜 대신 ID로 로드, 변환, 파일 구조 및 기타 흥미로운 사항에 대해.
• Apache Airflow에서 종속성 관리 - 지나갈 때만 언급한 작업 및 트리거 규칙의 종속성.
• 공기 흐름: DAG가 일정보다 훨씬 뒤처지는 경우 - 스케줄러에서 일부 "의도한 대로 작동"을 극복하고 손실된 데이터를 로드하고 작업의 우선 순위를 지정하는 방법.
• Apache Airflow에 유용한 SQL 쿼리 — Airflow 메타데이터에 대한 유용한 SQL 쿼리.
• Apache Airflow로 워크플로 개발 시작하기 - 맞춤형 센서 생성에 대한 유용한 섹션이 있습니다.
• Presto 및 Airflow를 사용하여 AWS에서 Fetchr 데이터 과학 인프라 구축 — 데이터 과학을 위해 AWS에서 인프라를 구축하는 것에 대한 흥미로운 짧은 메모입니다.
• Airflow DAG를 디버깅할 때 확인해야 할 7가지 일반적인 오류 - 일반적인 실수(누군가 여전히 지침을 읽지 않는 경우).
• Apache Airflow를 사용하여 비밀번호 저장 및 액세스 - Connections만 사용할 수 있지만 사람들이 비밀번호를 저장하는 방법에 대해 웃으세요.
• Zen of Python과 Apache Airflow - 암시적 DAG 포워딩, 함수에서 컨텍스트 던지기, 다시 종속성에 대해, 작업 실행 건너뛰기에 대해.
• 기류: 잘 알려지지 않은 팁, 요령 및 모범 사례 - 이용에 관하여 default arguments и params 템플릿, 변수 및 연결에서.
• Airflow 스케줄러 프로파일링 -기획자가 에어플로우 2.0을 어떻게 준비하고 있는지에 대한 이야기.
• docker-compose에서 3개의 Celery 작업자가 있는 Apache Airflow - 클러스터 배포에 대한 약간 오래된 기사 docker-compose.
• 4 Airflow 컨텍스트를 사용한 템플릿 작업 - 템플릿 및 컨텍스트 전달을 사용하는 동적 작업.
• Airflow의 오류 알림 — 메일 및 Slack을 통한 표준 및 사용자 지정 알림.
• 기류 워크숍: 목발이 없는 복잡한 DAG - 분기 작업, 매크로 및 XCom.

그리고 기사에 사용된 링크:

• 매크로 참조 - 템플릿에서 사용할 수 있는 자리 표시자.
• 일반적인 함정 - 공기 흐름 — dag를 만들 때 흔히 저지르는 실수.
• puckel/docker-airflow: 도커 아파치 공기 흐름 - docker-compose 실험, 디버깅 등을 위해.
• python-telegram-bot/python-telegram-bot: 거부할 수 없는 래퍼를 만들었습니다. — Telegram REST API용 Python 래퍼.

출처 : habr.com