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

안녕하세요, Habr! 이 글에서는 기업 DWH나 DataLake 인프라에서 일괄 데이터 처리 프로세스를 개발하는 데 유용한 도구 중 하나를 소개하고자 합니다. Apache Airflow(이하 Airflow)에 대해 알아보겠습니다. Habr에서는 이 도구에 대한 관심이 부족하지만, 본 글에서는 ETL/ELT 프로세스 스케줄러를 선택할 때 적어도 Airflow는 고려해 볼 만한 가치가 있다는 점을 강조하고자 합니다.

이전에 Tinkoff Bank에서 근무할 때 DWH를 주제로 여러 편의 기사를 썼습니다. 지금은 Mail.Ru 그룹 팀에 합류하여 게임 분야의 데이터 분석 플랫폼을 개발하고 있습니다. 새로운 소식과 흥미로운 솔루션이 나오는 대로, 저와 저희 팀은 여기에서 데이터 분석 플랫폼에 대해 이야기해 보겠습니다.

프롤로그

자, 시작해 볼까요. Airflow란 무엇일까요? 라이브러리(또는 라이브러리 세트) 워크플로 개발, 계획 및 모니터링을 위한 기능입니다. Airflow의 주요 기능: Python 코드를 사용하여 프로세스를 설명(개발)합니다. 이는 프로젝트 및 개발 관리에 많은 이점을 제공합니다. 실제로 (예를 들어) ETL 프로젝트는 Python 프로젝트일 뿐이며, 인프라 기능, 팀 규모 및 기타 요구 사항을 고려하여 원하는 대로 구성할 수 있습니다. 도구 측면에서는 모든 것이 간단합니다. 예를 들어 PyCharm + Git을 사용해 보세요. 훌륭하고 매우 편리합니다!

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

DAG

DAG는 특정 일정에 따라 엄격하게 정의된 순서대로 수행하려는 작업을 의미적으로 통합한 것입니다. Airflow는 DAG 및 기타 엔티티 작업을 위한 편리한 웹 인터페이스를 제공합니다.

DAG는 다음과 같습니다.

개발자는 DAG를 설계할 때 DAG 내부 작업이 구축될 연산자 집합을 정의합니다. 여기서 또 다른 중요한 엔티티인 Airflow 연산자를 살펴보겠습니다.

연산자

연산자는 작업 인스턴스가 생성되는 기반이 되는 엔티티로, 작업 인스턴스 실행 중에 무슨 일이 일어날지 설명합니다. GitHub의 Airflow 릴리스 이미 사용할 수 있는 연산자 집합이 포함되어 있습니다. 예:

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

더욱 구체적인 연산자로는 DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator가 있습니다.

필요에 따라 연산자를 개발하여 프로젝트에서 사용할 수도 있습니다. 예를 들어, MongoDB에서 Hive로 문서를 내보내는 연산자인 MongoDBToHiveViaHdfsTransfer와 Hive에서 작업하는 여러 연산자를 만들었습니다. 클릭 하우스: CHLoadFromHiveOperator와 CHTableLoaderOperator. 실제로 기본 연산자를 기반으로 자주 사용되는 코드가 프로젝트에 추가되면 이를 새로운 연산자로 모아서 사용할 수 있습니다. 이렇게 하면 추가 개발이 간소화되고 프로젝트의 연산자 라이브러리가 채워집니다.

다음으로, 이러한 모든 작업 인스턴스를 완료해야 하며, 이제 플래너에 대해 이야기하겠습니다.

스케줄러

Airflow의 작업 스케줄러는 다음을 기반으로 구축됩니다. 셀러리셀러리는 큐를 구성하고 작업의 비동기 및 분산 실행을 지원하는 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 수준에서 지정된 풀은 작업 수준에서 재정의될 수 있습니다.
별도의 프로세스인 스케줄러는 Airflow의 모든 작업을 스케줄링하는 역할을 합니다. 실제로 스케줄러는 작업 실행 설정과 관련된 모든 메커니즘을 처리합니다. 작업이 실행되기 전에 여러 단계를 거칩니다.

1. DAG에서는 이전 작업이 완료되면 새로운 작업을 대기열에 추가할 수 있습니다.
2. 대기열은 작업 우선순위에 따라 정렬되며(우선순위도 제어 가능), 풀에 빈 슬롯이 있으면 작업을 작업에 넣을 수 있습니다.
3. 사용 가능한 셀러리 워커가 있으면 작업은 해당 셀러리 워커에게 전달되고, 작업에서 프로그래밍한 작업은 한 명 또는 여러 명의 연산자를 사용하여 시작됩니다.

충분히 간단합니다.

스케줄러는 모든 DAG 세트와 DAG 내의 모든 작업에서 작동합니다.

Scheduler가 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 코드이므로, 예를 들어 샤딩된 소스 작업 시 코드 양을 매우 편리하게 줄일 수 있습니다. 예를 들어, 세 개의 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를 업데이트하는 것만으로 샤드를 추가하거나 제거할 수 있습니다. 편리하죠!

더 복잡한 코드 생성을 사용할 수도 있습니다. 예를 들어 DB 형태의 소스를 사용하거나 테이블 구조를 정의하고, 테이블 작업 알고리즘을 정의하고, DWH 인프라의 기능을 고려하여 N개의 테이블을 스토리지에 로드하는 프로세스를 생성할 수 있습니다. 또는 목록 형태의 매개변수 작업을 지원하지 않는 API를 사용하는 경우, 이 목록에 따라 DAG(다중 그룹)에서 N개의 작업을 생성하고, API 요청의 병렬 처리를 풀로 제한하고, API에서 필요한 데이터를 스크래핑할 수 있습니다. 유연하게 활용할 수 있습니다!

저장소

Airflow는 자체 백엔드 저장소인 DB(MySQL 또는 Postgres 중 하나, 저희는 Postgres를 사용합니다)를 가지고 있으며, 여기에는 작업 상태, DAG, 연결 설정, 전역 변수 등이 저장됩니다. Airflow의 저장소는 매우 단순하며(약 20개의 테이블) 자체 프로세스를 구축하려는 경우 편리합니다. 저는 Informatica 저장소에 100500개의 테이블이 있었던 것으로 기억하는데, 쿼리를 작성하기 전에 이를 이해하는 데 오랜 시간이 걸렸습니다.

모니터링

저장소의 단순성을 고려하면, 사용자에게 편리한 작업 모니터링 프로세스를 구축할 수 있습니다. Zeppelin의 노트북을 사용하여 작업 상태를 확인합니다.

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

Airflow 코드는 오픈 소스이므로 Telegram에 알림 기능을 추가했습니다. 실행 중인 모든 작업 인스턴스에서 오류가 발생하면 전체 개발 및 지원팀이 참여하는 Telegram 그룹에 스팸 메시지를 보냅니다.

필요한 경우 Telegram을 통해 신속한 답변을 받고, Zeppelin을 통해 Airflow의 전반적인 작업 상황을 파악할 수 있습니다.

전체로

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

현재 Airflow는 매일 실행되고 있습니다. 약 6,5천 개의 작업. 이 둘은 본질적으로 매우 다릅니다. 다양하고 매우 구체적인 소스에서 데이터를 메인 DWH로 로드하는 작업, 메인 DWH 내부에서 쇼케이스를 계산하는 작업, 빠른 DWH에 데이터를 게시하는 작업 등 매우 다양한 작업들이 있는데, Airflow는 이 모든 작업을 매일 처리합니다. 숫자로 이야기하자면, 2,3 천명 DWH(Hadoop) 내에서 다양한 복잡성을 지닌 작업의 ELT, 대략 2,5개의 데이터베이스 출처에 따르면, 이것은 팀입니다 4명의 ETL 개발자DWH 내부에서의 ETL 데이터 처리와 DWH 내부에서의 ELT 데이터 처리로 구분되며 물론 더 많은 한 명의 관리자서비스 인프라를 담당하는 기관입니다.

앞으로의 계획

프로세스 수는 필연적으로 증가하며, Airflow 인프라 측면에서 우리가 할 가장 중요한 일은 확장입니다. Airflow 클러스터를 구축하고, Celery Worker를 위한 레그(leg)를 할당하고, 작업 스케줄링 프로세스와 저장소를 갖춘 복제 헤드(duplicate head)를 만들고자 합니다.

커튼콜

물론, 이게 제가 Airflow에 대해 말씀드리고 싶은 전부는 아니지만, 핵심만 간략하게 설명하려고 노력했습니다. 식욕은 먹으면 따라오는 법이죠. 한번 드셔 보세요. 분명 마음에 드실 거예요 🙂

출처 : habr.com