Apache Airflow DAG 작성과 예약: 데이터 파이프라인의 스케줄링 마법

Apache Airflow에서 DAG(Directed Acyclic Graph)를 작성하고 예약하는 것은 데이터 파이프라인을 정의하고 관리하는 핵심적인 단계입니다. 이번 글에서는 Airflow에서 DAG를 작성하고 작업을 예약하는 방법에 대해 자세히 알아보겠습니다.

1. DAG 정의:

1.1 Python 스크립트 작성:

• DAG를 정의하려면 Python 스크립트를 작성해야 합니다. 이 스크립트는 각 작업을 DAG 객체에 추가하고 작업 간의 의존성을 설정하는 역할을 합니다.

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

# DAG 정의
dag = DAG(
    'my_dag',
    default_args={
        'owner': 'user',
        'start_date': datetime(2022, 1, 1),
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    schedule_interval='@daily',
)

1.2 작업 추가:

• DAG에 추가할 각 작업은 PythonOperator 등의 Operator를 사용하여 정의합니다. 작업 간의 의존성은 set_downstream(), set_upstream() 등의 메서드를 사용하여 설정합니다.

# 작업 정의
def my_python_function():
    print("Hello, Airflow!")

task1 = PythonOperator(
    task_id='task1',
    python_callable=my_python_function,
    dag=dag,
)

---

2. 작업 예약:

2.1 schedule_interval 설정:

• DAG에 포함된 작업들은 일정 주기로 예약됩니다. 이를 설정하기 위해 schedule_interval 매개 변수를 사용합니다. cron 표현식을 사용하여 정밀한 예약 주기를 설정할 수 있습니다.

dag = DAG(
    'my_dag',
    # ...
    schedule_interval='0 0 * * *',  # 매일 자정에 실행
)

2.2 시작 날짜 및 재시도 설정:

• DAG의 시작 날짜와 작업 재시도 관련 설정도 고려해야 합니다. default_args 매개 변수를 통해 이러한 설정을 추가할 수 있습니다.

dag = DAG(
    'my_dag',
    default_args={
        'owner': 'user',
        'start_date': datetime(2022, 1, 1),
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    # ...
)

---

3. DAG 테스트 및 실행:

3.1 DAG 구조 확인:

• 작성한 DAG의 구조를 시각적으로 확인하려면 Airflow UI를 통해 DAG 그래프를 살펴볼 수 있습니다.

3.2 DAG 실행:

• 작성한 DAG가 예약된 시간에 실행되도록 하려면 Airflow 스케줄러를 실행하고 DAG를 트리거할 수 있습니다.

airflow scheduler

---

4. 결론: Apache Airflow를 사용하여 DAG를 작성하고 예약하는 것은 데이터 파이프라인을 정의하고 자동화하는 핵심적인 부분입니다. 작업 간의 의존성을 설정하고 일정 주기로 작업을 예약함으로써 데이터 엔지니어 및 데이터 과학자는 효율적으로 데이터 워크플로우를 관리할 수 있습니다. Airflow의 강력한 스케줄링 기능을 활용하여 데이터 파이프라인의 안정성과 신뢰성을 높이고, 자동화된 환경에서 작업을 실행하여 생산성을 향상시킬 수 있습니다.