Airflow concepts (주요 용어)

Airflow concepts (주요 용어)

Concepts

DAG(direct Acyclic Graph)

방향성 비순항 그래프. DAG에 연결된 화살표 방향 순서대로 작업이 실행되고, 분기실행과 병렬 실행이 가능하다.

https://airflow.apache.org/docs/apache-airflow/stable/concepts/dags.html#

TASK

하나의 작업단위를 Task 라 부른다. 여러개의 TASK 가 합쳐저 DAG를 구성한다.

Task 간에는 <<, >> 기호를 사용하여 순서를 지정하는 것이 가능하다

Task는 밑에 설명할 Operator 들로 구성할수 있으며, Python Code 를 실행 시킬수 있는 PythonOperator와 Bash 를 실행할수있는 BashOperator, HTTP 요청을 할수 있는 SimpleHttpOperator등 여러 Operator를 제공한다.

Operator

Operator는 앞서 설명한 Task를 구성하기 위한 Airflow class 이다. Airflow 에서는 여러 Operator를 제공하고 상황에 따라 알맞는 Operator를 사용하면 된다.

Airflow operator 종류

Jinsa Template

Jinja 템플릿을 활용하여 Airflow 를 사용할 수 있다.

Sensor

센서는 정확히 한 가지 작업을 수행하도록 설계된 특수한 유형의 오퍼레이터 입니다.( 시간기반일수도있고, 파일이나 외부이벤트를 기다릴수도 있다.)

센서는 오퍼레이터와 마찬가지로 하나의 Task로 사용할 수 있는데 특정 조건이 채워지기를 기다리면 조건을 만족하는 경우 이후 Task로 넘어가게 하는 역할 이다.

Sensors - Airflow Documentation

센서는 주로 유휴 상태이기 때문에 세 가지 다른 실행 모드가 있으므로 사용에 있어 좀 더 효율적일 수 있습니다.

• poke(기본값): Sensor는 전체 런타임 동안 작업자 슬롯을 차지합니다.
• reschedule: Sensor는 검사 중일 때만 작업자 슬롯을 차지하며 검사 사이에 설정된 시간 동안 휴면합니다.
• smart sensor: 이 Sensor의 모든 실행을 일괄 처리하는 중앙 집중식 단일 버전이 있습니다.

poke및 모드 는 reschedule센서를 인스턴스화할 때 직접 구성할 수 있습니다. 일반적으로 둘 사이의 절충점은 대기 시간입니다. 1초마다 확인하는 것은 poke모드에 있어야 하고, 1분마다 확인하는 것은 reschedule모드에 있어야 합니다.

Pool

일부 시스템은 너무 많은프로세스가 동시에 처리될때 서비스가 종료될수있다. 풀을 사용하여 임의의 작업 집행에 대한 병령실행 처리를 제한 할 수 있다. 풀 목록은 풀에 이름을 지정하고 작업자 슬롯 수를 할당하여 UI 에서 관리 한다. 그런 다음에 작업을 생성할때 매개변수를 사용하여 작업을 기존 풀중에 하나와 연결 할 수 있다.

UI (Menu -> Admin -> Pools) 에서 생성 가능

aggregate_db_message_job = BashOperator(
    task_id="aggregate_db_message_job",
    execution_timeout=timedelta(hours=3),
    pool="ep_data_pipeline_db_msg_agg",
    bash_command=aggregate_db_message_job_cmd,
    dag=dag,
)
aggregate_db_message_job.set_upstream(wait_for_empty_queue)

Xcom

Xcom은 Airflow 의 task 간에 값을 전달하기 위해 사용된다. pull, push 방식으로 데이터가 공유된다. valiable과 같이 key-value 형식으로 사용된다.

주의할 점은 같은 DAG 안에서만 공유되며 다른 DAG에서도 공유하려면 Valiable과 같은 다른 방식을 사용하여야 한다.

Jinja template 을 사용한 XCom예제

Airflow XCOM 예제

Variable

변수는 Airflow 의 런타임 구성 개념입니다. 즉, 전역적이며 작업에서 쿼리할 수 있고, Airflow의 사용자 인터페이스를 통해 쉽게 설정하거나 JSON 파일로 대량 업로드할 수 있는 일반 키/값 저장소 입니다.

이것을 사용 하려면 변수모델을 가져오고 호출하기만 하면 된다.

Variable 예제

Airflow Variable 예제

Connection

Airflow는 종종 데이터를 다른 시스템으로 가져오고 보내는데 사용됨으로 외부 시스템과 통신되는데 커넥션은 사용되는 자격증명을 저장하기 위한 연결개념입니다.

연결은 기본적으로 사용자 이름및 암호, 호스트이름과 같은 매개변수 세트와 연결하는 시스템 유형 및 conn_id 라는 고유한 이름입니다.

UI, 또는 CLI 를 통해 관리가 가능 합니다.

자신의 코드에서 직접 Connections를 사용할 수 있습니다. Hooks 를 통해 사용하거나 템플릿에서도 사용할 수 있다.

Airflow connection 예제

Hook

훅은 외부플랫폼으로 빠르고 쉽게 연결할 수 있는 고수준의 인터페이스로, API에 도달하거나 특수라이브러리를 사용하는 저수준 코드를 작성할 필요 없이 빠르고 쉽게 외부 플랫폼과 대화 할수있다.

Hive , S3, MySQL, HDFS 등 여러 다른 플랫폼에 접근할수 있는 다양한 Hook 을 제공한다.

자격증명을 수집하기위해 Connections와 통합되며, 많은경우 기본 conn_id가 있다. 예를들어 PostgresHook 은 사용자가 값을 전달하지 않으면 conn_id 가 postgres_default인 연결을 자동으로 찾습니다.