템플릿, 멱등성, 백필

Airflow로 데이터 파이프라인 구축하기

Volker Janz

Senior Developer Advocate at Astronomer

Airflow의 Jinja 템플릿

@task.bash
def export_data():
    return "cp /data/sales.csv /export/sales_{{ ds }}.csv"

{{ ds }}는 논리적 날짜를 YYYY-MM-DD 형식으로 렌더링합니다
실행마다 고유한 날짜가 적용되므로, 동일한 태스크도 실행마다 다른 결과를 생성합니다
bash_command, sql 등 템플릿 적용 가능한 필드에서만 동작합니다

멱등성 문제

1회 실행 (3월 31일):

3월 31일 판매 데이터 INSERT
결과: 3개 행

2회 실행 (3월 31일 재실행):

3월 31일 판매 데이터 다시 INSERT
결과: 6개 행 (중복 발생!)

재실행 시 중복 행 발생

삭제 후 삽입 패턴

staged_rows = SQLExecuteQueryOperator(
    task_id="get_staged_rows",
    conn_id="duckdb_default",
    sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
)

load = SQLInsertRowsOperator(
    task_id="load_sales",
    conn_id="duckdb_default",
    table_name="sales",
    columns=["date", "product", "amount"],
    preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
    rows=staged_rows.output,
)

UPSERT 또는 MERGE 방식도 사용할 수 있습니다

삭제 후 삽입

백필

과거 날짜 범위를 재처리합니다
Airflow는 논리적 날짜마다 DAG 실행을 생성합니다
멱등성과 함께 사용하면 백필을 안전하게 실행할 수 있습니다

백필은 스케줄에 따라 달라짐

UI에서 백필 실행하기

DAG를 트리거하고, Backfill을 선택한 뒤 From 및 To 날짜를 설정합니다
Missing Runs, Missing and Errored Runs, All Runs 중 실행 유형을 선택할 수 있습니다
Max Active Runs로 병렬성을 설정하고 실행 순서를 변경할 수 있습니다
Run Backfill을 클릭하면 프로세스가 시작되고 실행이 생성됩니다

Airflow 백필 UI

연습해 봅시다!

Airflow로 데이터 파이프라인 구축하기