[Airflow] Airflow & DAG 설명

Apache Airflow

 

Apache Airflow는 데이터 파이프라인을 관리하고 스케줄링하기 위한 오픈 소스 플랫폼이다. Airflow는 파이썬 코드를 이용해 파이프라인을 구현할 수 있기에 파이썬 언어로 구현할 수 있는 대부분의 방법을 사용하여 여러 커스텀 파이프라인을 만들 수 있다. 또한 쉽게 확장 가능하고 다양한 시스템과 통합이 가능하다. 수많은 스케줄링 기법으로 파이프라인을 정기적으로 실행하고 점진적 처리가 가능하고 오픈 소스라는 장점이 있기 때문에 많은 기업에서 Airflow를 사용하고 있다.

(데이터 파이프라인은 데이터 처리 작업을 조직하고 실행하기 위한 일련의 단계 및 프로세스)

Apache Airflow의 특징

• 스케줄링과 모니터링
  • Airflow는 작업을 스케줄링하고 감시하는 데 사용됨
  • 예약된 작업은 DAG (Directed Acyclic Graph)라는 방향성이 있는 비순환 그래프로 정의됨
• 유연한 작업 정의
  • Python으로 작성된 코드를 사용하여 작업을 정의하므로 매우 유연
  • 사용자는 다양한 유형의 작업 및 데이터 처리 작업을 정의할 수 있음
• 모듈성과 재사용성
  • Airflow에서는 작업을 재사용 가능한 독립적인 유닛으로 정의할 수 있음
  • 이는 코드의 모듈성을 증가시키고 유지 보수를 용이하게 함
• 동적 확장성
  • Airflow는 동적으로 작업을 확장할 수 있는 기능을 제공
  • 새로운 작업이나 기능을 추가하거나 수정할 때 시스템을 중지시키지 않고도 변경 사항을 반영할 수 있음

 

DAG (Directed Acyclic Graph)

DAG는 Directed Acyclic Graph의 약자로, 방향성이 있는 비순환 그래프를 나타낸다. Airflow에서 DAG는 작업의 흐름이나 의존성을 정의하는데 사용되는데, 여러 작업들 간의 실행 순서와 의존성을 표현하는 그래프라고 보면 된다.

Airflow에서 DAG는 Python 스크립트로 정의되며, 이 스크립트는 작업들 간의 의존성 및 실행 스케줄을 명시한다. DAG 정의는 주로 다음과 같은 구성 요소로 이루어진다.

 

• DAG 객체 : DAG 클래스의 인스턴스로, 작업의 흐름과 스케줄을 정의
• 작업 (Task) : DAG 내에서 수행되어야 하는 각각의 단위 작업. 작업은 PythonOperator, BashOperator, Python 함수 등으로 정의.
• 의존성 (Dependencies) : 작업 간의 의존성. 즉, 어떤 작업은 다른 작업이 성공적으로 완료된 후에만 실행될 수 있도록 설정할 수 있음.
• 스케줄 (Schedule) : DAG 내의 작업이 실행되는 주기적인 스케줄을 정의. 예를 들어, 매일, 매주 특정 요일 등으로 스케줄할 수 있음.

 

코드 예시

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

# DAG 정의
dag = DAG(
    'my_dag',
    description='My example DAG',
    schedule_interval=timedelta(days=1),  # 매일 실행
    start_date=datetime(2023, 1, 1),
    catchup=False,  # 과거 실행에서 누락된 작업을 재실행하지 않음
)

# 작업 정의
def task1():
    print("Task 1")

def task2():
    print("Task 2")

# DAG에 작업 추가
t1 = PythonOperator(
    task_id='task1',
    python_callable=task1,
    dag=dag,
)

t2 = PythonOperator(
    task_id='task2',
    python_callable=task2,
    dag=dag,
)

# 의존성 정의
t1 >> t2

• 위 예제에서 t2는 t1이 성공적으로 완료된 후에만 실행됨
• 'schedule_interval'은 DAG가 얼마나 자주 실행될지를 정의
• 'start_date'는 DAG의 최초 실행 날짜를 정의