Airflow 101: 첫 워크플로우 만들기

Apache Airflow의 세계에 오신 것을 환영해요! 이 튜토리얼에서는 Airflow의 핵심 개념을 안내하며, 첫 DAG을 작성하는 방법을 이해하도록 도와드려요. Python에 익숙하든 이제 막 시작하든, 여정을 즐겁고 간단하게 만들어 드릴게요.

출처: 문서

본문

DAG이란 무엇인가요?

핵심적으로 DAG은 그들의 관계와 의존성을 반영하도록 구성된 task들의 모음이에요. 작업 흐름의 지도와 같아서 각 task가 다른 task와 어떻게 연결되는지 보여줘요. 조금 복잡하게 느껴진다 해도 걱정하지 마세요. 단계별로 쪼개서 설명할게요.

예시 파이프라인 정의

간단한 파이프라인 정의 예시로 시작해 볼게요. 처음엔 부담스러워 보일 수 있지만, 각 줄을 자세히 설명할 거예요.

airflow/example_dags/tutorial.py[source]

import textwrap
from datetime import datetime, timedelta

# Operators; we need this to operate!
from airflow.providers.standard.operators.bash import BashOperator

# The DAG object; we'll need this to instantiate a DAG
from airflow.sdk import DAG
with DAG(
    "tutorial",
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function, # or list of functions
        # 'on_success_callback': some_other_function, # or list of functions
        # 'on_retry_callback': another_function, # or list of functions
        # 'sla_miss_callback': yet_another_function, # or list of functions
        # 'on_skipped_callback': another_function, #or list of functions
        # 'trigger_rule': 'all_success'
    },
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        depends_on_past=False,
        bash_command="sleep 5",
        retries=3,
    )
    t1.doc_md = textwrap.dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](https://imgs.xkcd.com/comics/fixing_problems.png)
    **Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    templated_command = textwrap.dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id="templated",
        depends_on_past=False,
        bash_command=templated_command,
    )

    t1 >> [t2, t3]

DAG 정의 파일 이해하기

Airflow Python 스크립트를 DAG의 구조를 코드로 배치하는 구성 파일이라고 생각하세요. 여기서 정의하는 실제 task들은 다른 환경에서 실행되므로, 이 스크립트는 데이터 처리용이 아니에요. 주요 역할은 DAG 객체를 정의하는 것이며, Dag File Processor가 변경사항을 정기적으로 확인하므로 빠르게 평가되어야 해요.

모듈 가져오기 (Importing Modules)

시작하려면 필요한 라이브러리를 가져와야 해요. 이는 모든 Python 스크립트에서 전형적인 첫 단계예요.

airflow/example_dags/tutorial.py[source]

import textwrap
from datetime import datetime, timedelta

# Operators; we need this to operate!
from airflow.providers.standard.operators.bash import BashOperator

# The DAG object; we'll need this to instantiate a DAG
from airflow.sdk import DAG

Python과 Airflow가 모듈을 어떻게 처리하는지 자세한 내용은 Modules Management를 확인하세요.

기본 인자 설정 (Setting Default Arguments)

Dag과 그 task를 만들 때 각 task에 직접 인자를 전달하거나, 딕셔너리로 기본 파라미터 세트를 정의할 수 있어요. 후자의 접근 방식이 보통 더 효율적이고 깔끔해요.

airflow/example_dags/tutorial.py[source]

# These args will get passed on to each operator
# You can override them on a per-task basis during operator initialization
default_args={
    "depends_on_past": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
    # 'wait_for_downstream': False,
    # 'execution_timeout': timedelta(seconds=300),
    # 'on_failure_callback': some_function, # or list of functions
    # 'on_success_callback': some_other_function, # or list of functions
    # 'on_retry_callback': another_function, # or list of functions
    # 'sla_miss_callback': yet_another_function, # or list of functions
    # 'on_skipped_callback': another_function, #or list of functions
    # 'trigger_rule': 'all_success'
},

BaseOperator의 파라미터를 더 깊이 알고 싶다면 airflow.sdk.BaseOperator 문서를 살펴보세요.

DAG 만들기

다음으로 task들을 담을 DAG 객체를 만들어야 해요. DAG의 고유 식별자로 알려진 dag_id를 제공하고, 방금 정의한 기본 인자를 지정할 거예요. 또한 DAG이 매일 실행되도록 스케줄을 설정할 거예요.

airflow/example_dags/tutorial.py[source]

with DAG(
    "tutorial",
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function, # or list of functions
        # 'on_success_callback': some_other_function, # or list of functions
        # 'on_retry_callback': another_function, # or list of functions
        # 'sla_miss_callback': yet_another_function, # or list of functions
        # 'on_skipped_callback': another_function, #or list of functions
        # 'trigger_rule': 'all_success'
    },
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:

Operators 이해하기

Operator는 Airflow에서 작업의 단위를 나타내요. 워크플로우의 빌딩 블록으로, 어떤 task가 실행될지 정의할 수 있게 해줘요. 많은 작업에 operators를 사용할 수 있지만, Airflow는 더 파이써닉하게 워크플로우를 정의하는 TaskFlow API도 제공해요. 이는 나중에 다룰 거예요.

모든 operators는 Airflow에서 task를 실행하는 데 필요한 필수 인자를 포함하는 BaseOperator에서 파생돼요. 인기 있는 operators로는 PythonOperator, BashOperator, KubernetesPodOperator가 있어요. 이 튜토리얼에서는 간단한 bash 명령을 실행하기 위해 BashOperator에 초점을 맞출 거예요.

Task 정의하기

Operator를 사용하려면 그것을 task로 인스턴스화해야 해요. Task는 DAG의 컨텍스트 내에서 operator가 어떻게 작업을 수행할지 결정해요. 아래 예시에서 BashOperator를 두 번 인스턴스화해 두 개의 다른 bash 스크립트를 실행해요. task_id는 각 task의 고유 식별자 역할을 해요.

airflow/example_dags/tutorial.py[source]

t1 = BashOperator(
    task_id="print_date",
    bash_command="date",
)

t2 = BashOperator(
    task_id="sleep",
    depends_on_past=False,
    bash_command="sleep 5",
    retries=3,
)

operator 고유 인자(예: bash_command)와 BaseOperator에서 상속된 공통 인자(예: retries)를 혼합하는 방식을 눈여겨보세요. 이 접근 방식은 코드를 단순화해요. 두 번째 task에서는 retries 파라미터를 덮어써 3으로 설정하기도 했어요.

task 인자의 우선순위는 다음과 같아요:

명시적으로 전달된 인자
default_args 딕셔너리의 값
operator의 기본값 (가능한 경우)

Note

모든 task는 task_id와 owner 인자를 포함하거나 상속해야 해요. 그렇지 않으면 Airflow가 오류를 발생시켜요. 다행히 새 Airflow 설치에서는 owner가 airflow로 기본 설정되므로, 주로 task_id만 설정하면 돼요.

Templating에 Jinja 사용하기

Airflow는 Jinja Templating의 힘을 활용해 내장 파라미터와 매크로에 접근할 수 있게 해줘요. 이 섹션에서는 Airflow templating의 기초를 소개하며, 흔히 사용되는 템플릿 변수인 {{ ds }}에 초점을 맞출 거예요. 이 변수는 오늘의 날짜 스탬프를 나타내요.

airflow/example_dags/tutorial.py[source]

templated_command = textwrap.dedent(
    """
{% for i in range(5) %}
    echo "{{ ds }}"
    echo "{{ macros.ds_add(ds, 7)}}"
{% endfor %}
"""
)

t3 = BashOperator(
    task_id="templated",
    depends_on_past=False,
    bash_command=templated_command,
)

templated_command가 {% %} 블록 안에 로직을 포함하고 {{ ds }} 같은 파라미터를 참조하는 것을 볼 수 있어요. 또한 bash_command='templated_command.sh'처럼 파일을 bash_command에 전달할 수도 있어 코드를 더 잘 구성할 수 있어요. 템플릿에서 사용할 자신만의 변수와 필터를 만들기 위해 user_defined_macros와 user_defined_filters를 정의할 수도 있어요. 커스텀 필터에 대한 자세한 내용은 Jinja Documentation을 참조하세요.

템플릿에서 참조할 수 있는 변수와 매크로에 대한 자세한 정보는 Templates reference를 읽어보세요.

Dag과 Task 문서 추가하기

DAG이나 개별 task에 문서를 추가할 수 있어요. DAG 문서는 현재 markdown을 지원하지만, task 문서는 plain text, markdown, reStructuredText, JSON, YAML이 될 수 있어요. DAG 파일 시작 부분에 문서를 포함하는 것은 좋은 관행이에요.

airflow/example_dags/tutorial.py[source]

t1.doc_md = textwrap.dedent(
    """\
#### Task Documentation
You can document your task using the attributes `doc_md` (markdown),
`doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
rendered in the UI's Task Instance Details page.
![img](https://imgs.xkcd.com/comics/fixing_problems.png)
**Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
"""
)

dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
dag.doc_md = """
This is a documentation placed anywhere
"""  # otherwise, type it like this

../_images/task_doc.png ../_images/dag_doc.png

의존성 설정하기 (Setting up Dependencies)

Airflow에서 task는 서로 의존할 수 있어요. 예를 들어 t1, t2, t3 task가 있다면 여러 방법으로 의존성을 정의할 수 있어요:

t1.set_downstream(t2)

# This means that t2 will depend on t1
# running successfully to run.
# It is equivalent to:
t2.set_upstream(t1)

# The bit shift operator can also be
# used to chain operations:
t1 >> t2

# And the upstream dependency with the
# bit shift operator:
t2 << t1

# Chaining multiple dependencies becomes
# concise with the bit shift operator:
t1 >> t2 >> t3

# A list of tasks can also be set as
# dependencies. These operations
# all have the same effect:
t1.set_downstream([t2, t3])
t1 >> [t2, t3]
[t2, t3] << t1

Airflow는 Dag에서 사이클을 감지하거나 의존성이 여러 번 참조되면 오류를 발생시킨다는 점을 유의하세요.

시간대 다루기 (Working with Time Zones)

시간대를 인식하는 Dag을 만드는 것은 간단해요. pendulum으로 시간대를 인식하는 날짜를 사용하기만 하면 돼요. 알려진 제한이 있는 표준 라이브러리 timezone은 사용하지 마세요.

요약 (Recap)

축하해요! 이제 DAG을 만들고, task와 그 의존성을 정의하고, Airflow에서 templating을 사용하는 기본적인 이해를 갖게 됐어요. 코드는 다음과 비슷해야 해요:

airflow/example_dags/tutorial.py[source]

import textwrap
from datetime import datetime, timedelta

# Operators; we need this to operate!
from airflow.providers.standard.operators.bash import BashOperator

# The DAG object; we'll need this to instantiate a DAG
from airflow.sdk import DAG
with DAG(
    "tutorial",
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function, # or list of functions
        # 'on_success_callback': some_other_function, # or list of functions
        # 'on_retry_callback': another_function, # or list of functions
        # 'sla_miss_callback': yet_another_function, # or list of functions
        # 'on_skipped_callback': another_function, #or list of functions
        # 'trigger_rule': 'all_success'
    },
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        depends_on_past=False,
        bash_command="sleep 5",
        retries=3,
    )
    t1.doc_md = textwrap.dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](https://imgs.xkcd.com/comics/fixing_problems.png)
    **Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    templated_command = textwrap.dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id="templated",
        depends_on_past=False,
        bash_command=templated_command,
    )

    t1 >> [t2, t3]

파이프라인 테스트하기

이제 파이프라인을 테스트할 차례예요! 먼저 스크립트가 성공적으로 파싱되는지 확인하세요. airflow.cfg에 지정된 Dags 폴더의 tutorial.py에 코드를 저장했다면 다음을 실행할 수 있어요:

python ~/airflow/dags/tutorial.py

스크립트가 오류 없이 실행된다면, 축하해요! DAG이 올바르게 설정된 거예요.

명령줄 메타데이터 검증

몇 가지 명령을 실행해 스크립트를 더 검증해 볼게요:

# initialize the database tables
airflow db migrate

# print the list of active Dags
airflow dags list

# prints the list of tasks in the "tutorial" Dag
airflow tasks list tutorial

# prints the graphviz representation of "tutorial" Dag
airflow dags show tutorial

Task 인스턴스와 Dag Run 테스트

지정된 logical date에 대해 특정 task 인스턴스를 테스트할 수 있어요. 이는 스케줄러가 특정 날짜와 시간에 task를 실행하는 것을 시뮬레이션해요.

Note

스케줄러는 task를 특정 날짜와 시간 에 대해(for) 실행하지, 반드시 그(at) 날짜/시간에 실행하는 것은 아니라는 점을 눈여겨보세요. logical date는 Dag run이 이름 붙여진 타임스탬프로, 보통 워크플로우가 작동 중인 기간의 끝 — 또는 Dag run이 수동으로 트리거된 시각에 해당해요.

Airflow는 이 logical date를 사용해 각 실행을 구성하고 추적해요. UI, 로그, 코드에서 특정 실행을 참조하는 방식이에요. UI나 API를 통해 Dag을 트리거할 때 특정 시점 기준으로 워크플로우를 실행하도록 자신의 logical date를 제공할 수 있어요.

# command layout: command subcommand [dag_id] [task_id] [(optional) date]

# testing print_date
airflow tasks test tutorial print_date 2015-06-01

# testing sleep
airflow tasks test tutorial sleep 2015-06-01

템플릿이 어떻게 렌더링되는지도 다음을 실행해 볼 수 있어요:

# testing templated
airflow tasks test tutorial templated 2015-06-01

이 명령은 상세 로그를 제공하고 bash 명령을 실행해요.

airflow tasks test 명령은 task 인스턴스를 로컬에서 실행하고 로그를 stdout으로 출력하며 데이터베이스에 상태를 추적하지 않는다는 점을 기억하세요. 이는 개별 task 인스턴스를 테스트하는 편리한 방법이에요.

마찬가지로 airflow dags test는 단일 Dag run을 로컬에서 실행하며, 전체 Dag을 테스트하는 데 유용해요. airflow tasks test와 달리 실제 Dag run을 만들고 메타데이터 데이터베이스에 task 상태를 기록하므로, 초기화된 데이터베이스와 Dags 폴더에서 Airflow가 직렬화할 수 있는 Dag이 필요해요. dag.test()로 Dags 테스트하기를 참고하세요.

다음은 무엇일까요? (What's Next?)

여기까지입니다! 첫 Airflow 파이프라인을 성공적으로 작성하고 테스트했어요. 여정을 계속하면서, 코드를 스케줄러가 실행 중인 저장소에 병합하는 것을 고려해 보세요. 그러면 DAG이 매일 트리거되고 실행될 수 있어요.

다음 단계를 위한 몇 가지 제안:

참고 (See also)

튜토리얼의 다음 단계로: TaskFlow API로 파이써닉한 DAG 만들기
Dags, Tasks, Operators 같은 Airflow 개념에 대한 자세한 설명: Core Concepts 섹션 탐색

Airflow 101: 첫 워크플로우 만들기