[Airflow] 첫 DAG 만들기

🔨 첫 DAG 만들기

---

DAG를 생성하기 위해 먼저 dags 폴더를 하나 만들고 tutorials.py 라는 DAG를 작성할 python 파일을 생성한다.

처음으로 만들어볼 DAG는 3개의 Task로 구성된다.

• Task1 : 현재 날짜와 시간을 문자열로 출력
• Task2 : 5초 동안 대기 상태 유지
• Task3 : "Hello Airflow!" 문자열을 5번 출력
• Task4 : Gmail로 작업 완료 메일 전송

DAG가 트리거되면 Task1이 가장 먼저 수행되고 Task2, Task3가 병렬로 실행되고 그 이후 Task4가 수행된다.

webserver에서 Graph 기능을 제공하기 때문에 아래와 같이 내가 만든 DAG가 어떤 순서로 동작하는지 시각적으로 확인할 수 있다.

🪄 Importing Modules

---

DAG 객체를 정의하기 위해 Python 스크립트 작성에 필요한 Module을 Import한다.

import textwrap
from datetime import datetime, timedelta

# DAG object; DAG 초기화를 위해 사용
from airflow.models.dag import DAG

# Operatorsl; DAG 동작을 위해 사용
from airflow.operators.bash import BashOperator

🔍 Default Arguments

DAG 작성에 필요한 Arguments에 대해서 알아보자.

• depends_on_past
  - 이전 실행의 성공 여부에 따라 현재 태스크 실행을 결정
  • True: 이전 실행이 성공해야만 현재 태스크가 실행
  • False: 이전 실행의 성공 여부와 상관없이 현재 태스크가 실행
• eamil
  - 태스크 실패나 재시도 시 알림을 받을 이메일 주소 목록을 지정
• email_on_failure
  - 태스크가 실패할 때 이메일 알림을 보낼지 여부를 결정
  • True: 태스크 실패 시 이메일 알림을 보낸다.
  • False: 태스크 실패 시 이메일 알림을 보내지 않는다.
• email_on_retry
  - 태스크가 재시도될 때 이메일 알림을 보낼지 여부를 설정
  • True: 태스크 재시도 시 이메일 알림을 보냄
  • False: 태스크 재시도 시 이메일 알림을 보내지 않는다.
• retires
  - 태스크 실패 시 재시도 횟수를 설정
• retry_delay
  - 태스크 실패 후 재시도 전 대기 시간을 설정
• queue
  - 태스크를 실행할 큐를 지정. Airflow에서 여러 큐를 사용할 수 있음.
• pool
  - 태스크가 속할 풀을 지정. 풀은 자원 할당을 제한하는 데 사용
• priority_weight
  - 태스크의 우선 순위를 지정. 숫자가 클수록 높은 우선 순위를 가짐
• end_date
  - 태스크가 실행될 수 있는 마지막 날짜를 지정
• wait_for_downstream
  - 모든 하위 태스크가 완료될 때까지 현태 태스크가 완료되지 않도록 설정
  • True: 하위 태스크가 모두 완료될 때까지 기다림
  • False: 기다리지 않음
• sla
  - 특정 시간 내에 태스크가 완료되지 않으면 SLA 경고를 발생시킴
• execution_timeout
  - 태스크가 지정된 시간 내에 완료되지 않으면 실패로 간주
• on_failure_callback
  - 태스크가 실패했을 때 콜백 함수를 지정
• on_success_callback
  - 태스크가 성공했을 때 실행할 콜백 함수를 지정
• `on_retry_callback
  - 태스크가 재시도될 때 실행할 콜백 함수를 지정
• sla_miss_callback
  - 태스크가 SLA를 충족하지 못했을 때 실행할 콜백 함수를 지정
• on_skipped_callback
  - 태스크가 건너뛰기 되었을 때 실행할 콜백 함수를 지정
• trigger_rule
  - 태스크가 실행될 트리거 규칙을 지정. 기본값은 all_success로, 모든 상위 태스크가 성공해야 실행
  • all_sucess: 모든 상위 태스크가 성공해야 실행
  • all_failed: 모든 상위 태스크가 실패해야 실행
  • all_done: 모든 상위 태스크가 완료되면 실행
  • one_success: 하나의 상위 태스크가 성공하면 실행
  • one_failed: 하나의 상위 태스크가 실패하면 실행
  • none_failed: 실패한 상위 태스크가 없으면 실행
  • none_skipped: 건너뛴 상위 태스크가 없으면 실행
  • dummy: 트리거 규칙을 무시하고 항상 실행

🖊️ Instantiate a DAG

"tutorial"이라는 이름의 DAG를 정의하고, 기본 인수, 스케줄 간격, 시작 날짜, 설명, 태그 등을 설정한다. DAG 내에 정의된 태스크들은 이 설정들을 기본으로 사용하게 된다.

이 DAG는 하루에 한 번 실행되며, 2021년 1월 1일 이후의 실행부터 스케줄링되며 과거의 스케줄된 실행은 무시된다.

• dag_id: DAG의 고유 식별자를 설정한다. DAG 이름은 Webserver, 로그에서 사용된다.
• default_args: DAG 객체 정의를 위한 arguments를 딕셔너리 형태로 전달한다.
• description: DAG의 설명을 설정한다.
• schedule_interval: DAG가 실행될 간격을 설정한다.
• start_date: DAG의 시작 날짜를 설정한다. 이 날짜 이후에 DAG가 스케줄링된다.
• catchup: DAG가 시작 날짜 이후에 실행되지 않은 기간 동안의 모든 실행을 완료할지 여부를 설정한다. False의 경우 과거의 스케줄된 실행을 무시한다.
• tags: DAG의 태그를 설정한다. Webserver에서 DAG를 검색하고 필터링하는 데 사용된다.

with DAG(
    "my_first_dag",
    # 이 인자들은 각 operator에 전달됨
    # operator 초기화 중에 각 task별로 오버라이딩이 가능
    default_args={
        "depends_on_past": False,
        "email": ["kimdong799@gmail.com"],
        "email_on_failure": False,
        "email_on_retry": 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,
        # 'sla': timedelta(hours=2),
        # '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:

중간에 DAG를 수정했다면 다음의 명령어를 실행하자

airflow db init
airflow scheduler -D

Operator

Operator는 워크플로우 내에서 단일 Task를 정의하는 구성 요소이다.
각 Operator는 특정 유형의 Task를 수행하기 위한 로직을 캡슐화한다.

Operator는 기본적으로 BaseOperator를 상속받으며 작업을 실행하기 위한 기본적인 인자들을 포함한다.

Operator의 종류

• BashOperator: Bash 명령어 또는 스크립트를 실행
• PythonOperator: Python 함수를 실행
• KubernetesPodOperator: Kubernetes Pod에서 작업을 실행
• SimpleHttpOperator: HTTP 요청을 보냄
• EmailOperator: 이메일을 보냄

Tasks

앞서 설명한 Operator를 이용하여 각 독립된 Task를 정의한다.
정의된 태스크의 실행 순서는 >> 기호를 이용하여 작성할 수 있다.

    # t1, t2 및 t3, t4는 operator를 인스턴스화하여 생성된 task의 예시
    task1 = BashOperator(
        task_id="print_date",
        bash_command="date",  # 현재 날짜와 시간을 문자열로 출력
    )

    task2 = BashOperator(
        task_id="sleep",
        # depends_on_past: False,
        bash_command="sleep 5",  # 5초 동안 대기 상태 유지
        retries=3,
    )

    # Task에 대한 문서를 작성
    task1.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__  # DAG시작 부분에 docstrin이 있는 경우
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # 그렇지 않은 경우 아래와 같이 작성
    templated_command = textwrap.dedent(
        """
        {% for i in range(5) %}
            echo "Hello Airflow!"
        {% endfor %}
        """
    )

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

    task4 = EmailOperator(
        task_id="send_email",
        to="kimdong799@gmail.com",  # 수신자 이메일 주소
        subject="Airflow Alert",
        html_content="""<h3>DAG Completed</h3><p>This is a test email sent from an Airflow DAG.</p>""",
    )

    task1 >> [task2, task3] >> task4

지금까지 정의한 DAG를 Webserver에서 실시간으로 모니터링 할 수 있어 굉장히 간편하다.

모든 Task가 정상적으로 수행되어 메일까지 도착한 것을 확인했다.

ref : https://airflow.apache.org/docs/apache-airflow/stable/tutorial/fundamentals.html

ref : https://stackoverflow.com/questions/51829200/how-to-set-up-airflow-send-email