Airflow는 일괄 데이터 처리 프로세스를 편리하고 빠르게 개발하고 유지 관리하기 위한 도구입니다.

안녕, 하브르! 이 기사에서는 예를 들어 기업 DWH 또는 DataLake의 인프라에서 일괄 데이터 처리 프로세스를 개발하기 위한 훌륭한 도구 중 하나에 대해 이야기하고 싶습니다. Apache Airflow(이하 Airflow)에 대해 이야기하겠습니다. 이는 Habré에 대한 관심을 부당하게 빼앗긴 것이며, 주요 부분에서는 ETL/ELT 프로세스에 대한 스케줄러를 선택할 때 적어도 Airflow를 살펴볼 가치가 있음을 확신시키려고 노력할 것입니다.

이전에 저는 Tinkoff Bank에서 근무할 때 DWH를 주제로 일련의 기사를 썼습니다. 이제 저는 Mail.Ru 그룹 팀의 일원이 되어 게임 분야의 데이터 분석을 위한 플랫폼을 개발하고 있습니다. 실제로 뉴스와 흥미로운 솔루션이 나오면 저희 팀과 저는 여기서 데이터 분석을 위한 플랫폼에 대해 이야기하겠습니다.

프롤로그

그럼 시작해 보겠습니다. 공기 흐름이란 무엇입니까? 이곳은 도서관(혹은 라이브러리 세트) 작업 프로세스를 개발, 계획 및 모니터링합니다. Airflow의 주요 기능: Python 코드는 프로세스를 설명(개발)하는 데 사용됩니다. 이는 프로젝트 및 개발을 구성하는 데 많은 이점이 있습니다. 본질적으로 (예를 들어) ETL 프로젝트는 Python 프로젝트일 뿐이며 인프라, 팀 규모 및 환경의 세부 사항을 고려하여 원하는 대로 구성할 수 있습니다. 다른 요구 사항. 도구적으로 모든 것이 간단합니다. 예를 들어 PyCharm + Git을 사용하세요. 훌륭하고 매우 편리합니다!

이제 Airflow의 주요 항목을 살펴보겠습니다. 그 본질과 목적을 이해함으로써 프로세스 아키텍처를 최적으로 구성할 수 있습니다. 아마도 주요 엔터티는 방향성 비순환 그래프(이하 DAG)일 것입니다.

DAG

DAG는 특정 일정에 따라 엄격하게 정의된 순서로 완료하려는 작업의 의미 있는 연결입니다. Airflow는 DAG 및 기타 항목 작업을 위한 편리한 웹 인터페이스를 제공합니다.

DAG는 다음과 같습니다.

개발자는 DAG를 설계할 때 DAG 내 작업이 구축될 연산자 세트를 배치합니다. 여기에서 또 다른 중요한 엔터티인 Airflow Operator를 살펴보겠습니다.

연산자

운영자는 작업 인스턴스가 생성되는 기반이 되는 엔터티로, 작업 인스턴스 실행 중에 발생하는 상황을 설명합니다. GitHub의 Airflow 릴리스 이미 사용할 수 있는 연산자 세트가 포함되어 있습니다. 예:

• BashOperator - bash 명령을 실행하는 연산자입니다.
• PythonOperator - Python 코드를 호출하기 위한 연산자입니다.
• EmailOperator — 이메일 전송을 위한 연산자입니다.
• HTTPOperator - http 요청 작업을 위한 연산자입니다.
• SqlOperator - SQL 코드 실행을 위한 연산자입니다.
• 센서는 이벤트(필요한 시간 도착, 필요한 파일의 모양, 데이터베이스의 라인, API의 응답 등)를 기다리는 연산자입니다.

DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator와 같은 보다 구체적인 연산자가 있습니다.

또한 자신의 특성에 따라 연산자를 개발하여 프로젝트에 사용할 수도 있습니다. 예를 들어 MongoDB에서 Hive로 문서를 내보내는 연산자인 MongoDBToHiveViaHdfsTransfer와 작업을 위한 여러 연산자를 만들었습니다. 클릭 하우스: CHLoadFromHiveOperator 및 CHTableLoaderOperator. 기본적으로 프로젝트에서 기본 문을 기반으로 작성된 코드를 자주 사용하면 이를 새 문으로 작성하는 것을 고려할 수 있습니다. 이렇게 하면 추가 개발이 단순화되고 프로젝트에서 연산자 라이브러리가 확장됩니다.

다음으로 이러한 모든 작업 인스턴스를 실행해야 하며 이제 스케줄러에 대해 설명하겠습니다.

스케줄러

Airflow의 작업 스케줄러는 다음을 기반으로 합니다. 셀러리. Celery는 대기열을 구성하고 작업의 비동기 및 분산 실행을 구성할 수 있는 Python 라이브러리입니다. Airflow 측에서는 모든 작업이 풀로 구분됩니다. 풀은 수동으로 생성됩니다. 일반적으로 그 목적은 소스 작업 작업 부하를 제한하거나 DWH 내 작업을 대표하는 것입니다. 풀은 웹 인터페이스를 통해 관리할 수 있습니다.

각 풀에는 슬롯 수에 제한이 있습니다. DAG를 만들 때 풀이 제공됩니다.

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

DAG 수준에서 정의된 풀은 작업 수준에서 재정의될 수 있습니다.
별도의 프로세스인 Scheduler가 Airflow의 모든 작업 예약을 담당합니다. 실제로 스케줄러는 실행 작업을 설정하는 모든 메커니즘을 다룹니다. 작업은 실행되기 전에 여러 단계를 거칩니다.

1. 이전 작업은 DAG에서 완료되었으며 새 작업이 대기열에 추가될 수 있습니다.
2. 대기열은 작업의 우선순위에 따라 정렬되며(우선순위도 제어 가능) 풀에 여유 슬롯이 있으면 작업을 실행할 수 있습니다.
3. 무료 작업자 셀러리가 있는 경우 작업이 해당 셀러리로 전송됩니다. 문제에 프로그래밍한 작업은 하나 또는 다른 연산자를 사용하여 시작됩니다.

충분히 간단합니다.

스케줄러는 모든 DAG 집합과 DAG 내의 모든 작업에서 실행됩니다.

스케줄러가 DAG 작업을 시작하려면 DAG가 일정을 설정해야 합니다.

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

미리 만들어진 사전 설정 세트가 있습니다: @once, @hourly, @daily, @weekly, @monthly, @yearly.

cron 표현식을 사용할 수도 있습니다.

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

실행 날짜

Airflow의 작동 방식을 이해하려면 DAG의 실행 날짜를 이해하는 것이 중요합니다. Airflow에서 DAG에는 실행 날짜 차원이 있습니다. 즉, DAG의 작업 일정에 따라 각 실행 날짜에 대해 작업 인스턴스가 생성됩니다. 그리고 각 실행 날짜에 대해 작업을 다시 실행할 수 있습니다. 예를 들어 DAG는 여러 실행 날짜에 동시에 작업할 수 있습니다. 이는 여기에 명확하게 표시됩니다.

불행하게도(또는 다행스럽게도 상황에 따라 다름) DAG의 작업 구현이 수정되면 조정 사항을 고려하여 이전 실행 날짜의 실행이 진행됩니다. 새로운 알고리즘을 사용하여 과거 기간의 데이터를 다시 계산해야 하는 경우에는 좋지만 결과의 재현성이 손실되기 때문에 좋지 않습니다(물론 Git에서 필요한 소스 코드 버전을 반환하고 무엇을 계산하도록 귀찮게 하는 사람은 아무도 없습니다). 필요한 방식으로 일회성이 필요합니다).

작업 생성

DAG의 구현은 Python의 코드이므로, 예를 들어 샤딩된 소스로 작업할 때 코드 양을 줄이는 매우 편리한 방법이 있습니다. 소스로 XNUMX개의 MySQL 샤드가 있고 각 샤드로 이동하여 일부 데이터를 가져와야 한다고 가정해 보겠습니다. 또한 독립적으로 그리고 병렬로. DAG의 Python 코드는 다음과 같습니다.

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

DAG는 다음과 같습니다.

이 경우 간단히 설정을 조정하고 DAG를 업데이트하여 샤드를 추가하거나 제거할 수 있습니다. 편안한!

예를 들어 데이터베이스 형태의 소스로 작업하거나 테이블 구조, 테이블 작업 알고리즘을 설명하고 DWH 인프라의 기능을 고려하여 프로세스를 생성하는 등 더 복잡한 코드 생성을 사용할 수도 있습니다. N개의 테이블을 스토리지에 로드하는 데 사용됩니다. 또는 예를 들어 목록 형식의 매개변수 작업을 지원하지 않는 API를 사용하여 이 목록에서 DAG에 N개의 작업을 생성하고 API의 요청 병렬 처리를 풀로 제한하고 스크랩할 수 있습니다. API에서 필요한 데이터. 유연한!

저장소

Airflow에는 작업 상태, DAG, 연결 설정, 전역 변수 등을 저장하는 자체 백엔드 저장소인 데이터베이스(MySQL 또는 Postgres일 수 있고 Postgres가 있음)가 있습니다. 여기서는 다음과 같이 말할 수 있습니다. Airflow의 저장소는 매우 간단하고(테이블 약 20개) 그 위에 자체 프로세스를 구축하려는 경우 편리합니다. 쿼리 작성 방법을 이해하기 전에 오랫동안 연구해야 했던 Informatica 리포지토리에 있는 100500개의 테이블을 기억합니다.

모니터링

리포지토리의 단순성을 고려하면 편리한 작업 모니터링 프로세스를 구축할 수 있습니다. 우리는 Zeppelin에서 메모장을 사용하여 작업 상태를 확인합니다.

이는 Airflow 자체의 웹 인터페이스일 수도 있습니다.

Airflow 코드는 오픈 소스이므로 Telegram에 알림을 추가했습니다. 작업의 각 실행 인스턴스는 오류가 발생하면 전체 개발 및 지원 팀이 구성된 Telegram의 그룹에 스팸을 보냅니다.

Telegram을 통해(필요한 경우) 즉각적인 응답을 받고, Zeppelin을 통해 Airflow의 작업에 대한 전반적인 그림을 받습니다.

전체로

Airflow는 기본적으로 오픈 소스이므로 기적을 기대해서는 안 됩니다. 효과적인 솔루션을 구축하기 위해 시간과 노력을 투자할 준비를 하십시오. 목표는 달성 가능합니다. 저를 믿으세요. 그만한 가치가 있습니다. 개발 속도, 유연성, 새로운 프로세스 추가 용이성 등이 마음에 드실 것입니다. 물론 프로젝트 구성, Airflow 자체의 안정성에 많은 관심을 기울여야 합니다. 기적은 일어나지 않습니다.

이제 Airflow가 매일 작동합니다. 약 6,5개 작업. 그들은 성격이 상당히 다릅니다. 다양하고 매우 구체적인 소스에서 기본 DWH로 데이터를 로드하는 작업이 있고, 기본 DWH 내부의 매장 전면을 계산하는 작업이 있으며, 빠른 DWH에 데이터를 게시하는 작업이 있고, 매우 다양한 작업이 있습니다. 매일 다 씹어 먹습니다. 숫자로 말하면 이렇다. 2,3 천명 DWH(Hadoop) 내에서 다양한 복잡성을 지닌 ELT 작업. 2,5개의 데이터베이스 소식통에 따르면 여기는 팀입니다. 4명의 ETL 개발자, 이는 DWH에서의 ETL 데이터 처리와 DWH 내부의 ELT 데이터 처리로 구분됩니다. 관리자 한 명, 서비스 인프라를 담당합니다.

앞으로의 계획

프로세스 수는 필연적으로 증가하고 있으며 Airflow 인프라 측면에서 우리가 할 주요 작업은 확장입니다. 우리는 Airflow 클러스터를 구축하고, Celery 작업자를 위한 한 쌍의 다리를 할당하고, 작업 일정 프로세스와 저장소를 갖춘 자체 복제 헤드를 만들고 싶습니다.

커튼콜

물론 이것이 Airflow에 대해 말하고 싶은 전부는 아니지만 주요 사항을 강조하려고 노력했습니다. 먹으면 식욕이 따라오는데 꼭 드셔보시면 좋아하실 것 같아요 :)

출처 : habr.com