Airflow 용어 및 기본 개념

Airflow 구성

지난 포스트에서 Airflow가 무엇인지 간단하게 알아보았다

이번 포스트에서는 Airflow에서 알아야 할 용어들과 기본 개념에 대해서 알아보도록 하자

DAG (Directed Acyclic Graph)

지향성 비순환 그래프로 지난 포스트에서도 간단하게 언급이 되었는데

python으로 작성하고 순서를 정해 하나의 workflow형식으로 동작한다

DAG는 아래와 같은 그림으로 표시할 수 있으며, 각 노드들은 task로 DAG가 실행되는 순서를 파악할 수 있다.

Airflow에서 DAG를 인식하기 위해서는 dags folder로 지정된 경로에 존재하는 python 파일로 "airflow" 혹은 "dag"라는 키워드가 포함되어 있어야 한다.

모든 python 파일을 인식시키기 위해선 DAG_DISCOVERY_SAFE_MODE를 disable로 설정하면 되지만 권장되는 사항은 아니다.

 

https://en.wikipedia.org/wiki/Directed_acyclic_graph

Task

하나의 작업 단위를 Task라고 하며 하나 또는 여러 개의 Task를 이용해 하나의 DAG를 생성하게 된다

Task 간에는 순서를 지정할 수 있으며, <<, >>를 통해 간단하게 지정하는 것이 가능하다

Task는 Operator로 만들 수 있으며, Python code를 실행시키기 위한 PythonOperator, Bash command를 실행시키기 위한 BashOperator 등이 제공된다.

with DAG('my_dag', start_date=datetime(2016, 1, 1)) as dag:
    task_1 = DummyOperator('task_1')
    task_2 = DummyOperator('task_2')

    # task_1을 실행한 후 task_2 실행
    task_1 >> task_2 # Define dependencies

Operator

Operator은 Task를 만들기 위해 사용되는 Airflow class인데 위에서 언급한 BashOperator, PythonOperator등을 제외한 외부 서비스들과 연동을 위한 Operator들이 제공되며 args를 이용해 어떻게 동작할지를 정의할 수 있다 

Operator의 종류는 아래의 링크에서 더 자세히 확인할 수 있다

airflow.apache.org/docs/stable/_api/airflow/operators/index.html

airflow.apache.org/docs/

제공된 Operator뿐만 아니라 Operator을 생성하는 것도 가능하다

Sensor

Sensor은 Operator과 마찬가지로 하나의 Task로 사용할 수 있는데 특정 조건이 채워지기를 기다리면 조건을 만족하는 경우 이후 Task로 넘어가게 하는 역할이다

예를 들어 특정 경로에 특정 파일이 존재해야 동작하는 workflow가 있는 경우 Sensor을 이용해 해당 파일이 있는지를 검사하고 다음 Task가 실행되도록 할 수 있다

Sensor의 종류에도 여러가지가 있는데 아래의 링크에서 확인이 가능하다

airflow.apache.org/docs/stable/_api/airflow/sensors/index.html

Pool

Airflow는 동시성 제어를 위해 Pool이라는 기능도 제공하는데 이는 Web UI에서 제어하며

Pool 안에서 지정된 슬롯 숫자만큼의 DAG만 동시에 실행 가능하도록 설정할 수 있다

자원이 한정적인 cluster환경에서 우선순위가 떨어지는 DAG를 사용할 때 이용하면 자원 관리를 좀 더 효율적으로 할 수 있다

Xcom

Xcom은 Airflow task간 데이터 전달을 위한 기능으로 push, pull 방식으로 데이터를 공유하는 기능을 말하는데

주의할 점은 한 DAG안에서만 공유된다는 점이다 (다른 DAG 간 공통된 값을 이용하려는 경우에는 Variable를 사용)

xcom_push, xcom_pull을 이용해 데이터를 전달하고 전달받는 방식이며 PythonOperator을 사용하는 경우에는 return 값이 자동으로 Xcom에 push 된다

Xcom에 대한 설명을 보고 DAG파일 안에서 데이터를 공유하는 것이 가능한데 Xcom이라는 기능이 왜 필요한지를 생각할 수 있는데, Airflow Worker을 여러 개 사용하는 경우 하나의 DAG가 동일한 Worker에서 동작하는 것을 보장할 수 없다 예를 들어, DAG안에 t1, t2 task가 존재하는 경우 t1은 server1에서 동작하고, t2는 server2에서 동작하게 되면 task 간 데이터를 공유하기 위해 Xcom을 사용하는 방법밖에 없는 것이다.

Variable

Variable는 위에서 언급한 것과 같이 Airflow에서 공통적으로 사용 가능한 변수들을 모아놓는 곳으로 Web UI에서 관리가 가능하다 key-value의 형식으로 저장되어 있으며 보안을 위해 password, secret, passwd, authorization, api_key, apikey, access_token 등의 키워드가 포함된 key를 가지는 경우 Web UI에서는 value가 보이지 않는 암호화된 key를 생성할 수 있다.

출처 : https://airflow.apache.org/docs/1.10.13/ui.html#variable-view 

Connection

Connecion은 외부 시스템과 연결하는 방식에 대한 정보를 저장해놓는 곳으로 Operator, Hook등에서 Connection의 정보를 사용한다. UI에서 설정이 가능하며, (Menu -> Admin -> Connections)에서 설정 및 수정 등이 가능

인증을 위한 username, password에 대한 설정과 host에 대한 설정, 사용 Hook, Operator에 따라 추가 옵션도 설정할 수 있다

Hook

Hook는 외부 플랫폼에 대한 인터페이스를 제공하는 것으로 Hive, S3, MySQL HDFS 등에 접근할 수 있는 다양한 Hook을 제공 Hook은 독립적으로 task가 될 수 없으며, Operator과 함께 사용된다.