이 챕터에서는 Apache Airflow의 구성 요소를 전반적으로 살펴보고, 각 구성 요소를 사용하는 방법과 이유를 배웁니다.

Apache Airflow 소개

DAG 개수 확인

첫 번째 DAG 실행하기

Airflow DAG

간단한 DAG 만들기

DAG과 Airflow 셸 사용하기

Airflow 명령어 살펴보기

Airflow에서 태스크 테스트하기

Airflow 웹 인터페이스

Airflow UI 탐색하기

Airflow UI로 DAG 살펴보기

Dag 오류

Airflow 소개

이제 Airflow DAG 구현의 기본을 배울 시간입니다. 실습 활동을 통해 오퍼레이터, 태스크, 스케줄링을 설정하고 배포하는 방법을 익혀 보세요.

Airflow 연산자

태스크 데코레이팅

@task.bash로 BashOperator 정의하기

태스크 실행 순서 정의하기

@task 더 활용하기

DAG 스케줄링

Python으로 DAG 스케줄 설정하기

Airflow 스케줄 해석하기

DAG 실행 문제 해결

XCom으로 태스크 간 데이터 전달하기

XCom으로 데이터 전달하기

XCom 변수

Airflow 센서

태스크 실행 순서 파악하기

FileSensor 만들기

실행이 멈춘 DAG 분석하기

Airflow에서 DAG 구성하기

이 챕터에서는 센서, 익스큐터 등 Airflow 구성 요소를 활용해 시간을 절약하는 방법과 함께 Airflow 워크플로를 모니터링하고 문제를 해결하는 방법을 배웁니다.

모니터링, 알림 및 콜백

콜백 함수 구현하기

SmtpNotifier 구현하기

로그에는 무엇이 있을까요?

Jinja를 활용한 템플릿 작성

Jinja로 템플릿 작성하기

템플릿으로 교체하기

Airflow 변수

정답은 Variable

Python에서 변수 읽기

변수 설정이 올바르지 않은 경우

Airflow 디버깅 및 문제 해결

Dag 문제 해결

from airflow.sdk import dag, task
#from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

from airflow.sdk import dag, task
from pendulum import datetime

@dag(
    dag_id="sample_dag",
    schedule="0 0 * * *",
    start_date=datetime(2023, 2, 20),
)
def sample_dag():

    @task()
    def sample():
        print("Running sample task...")

    sample()

sample_dag()

from airflow.sdk import dag, task
from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

누락된 DAG

Airflow 워크플로 유지 관리 및 모니터링

지금까지 배운 내용을 모두 종합해 봅니다. 마지막 챕터에서는 학습한 모든 내용을 적용하여 프로덕션 수준의 워크플로를 Airflow에서 구축합니다.

트리거와 오류 허용

트리거 규칙 구현하기

재시도 추가하기

자식 Dag 트리거하기

브랜칭(Branching)

@task.branch 정의하기

브랜칭 오류 해결

HITL 연산자

사람의 승인 단계 추가하기

ApprovalOperator 태스크

프로덕션 파이프라인 구축하기

판매 ETL 파이프라인 설정하기

월별 분기 추가하기

축하합니다!

DAG 로직 제어

Apache Airflow 3.1.6 업데이트 버전 - 데이터를 정해진 일정에 맞춰 전달하는 작업은 번거로울 수 있습니다. 스크립트를 작성하고, 복잡한 cron 작업을 추가하고, 끊임없이 바뀌는 요구사항을 충족하기 위해 다양한 방법을 시도해야 하죠. 팀원들과 함께 작업할 때는 더욱 관리하기 어렵습니다. Apache Airflow를 사용하면 워크플로에 스케줄링, 오류 처리, 보고 기능을 추가해 이러한 번거로움을 해소할 수 있습니다. 이 과정에서는 Apache Airflow의 기본기를 탄탄히 다지고, 프로덕션 환경에서 복잡한 데이터 엔지니어링 파이프라인을 구현하는 방법을 배웁니다. 또한 방향성 비순환 그래프(DAG)를 활용하고, 데이터 엔지니어링 워크플로를 자동화하며, 데이터 엔지니어링 작업을 쉽고 반복 가능한 방식으로 구현하는 방법도 익힐 수 있습니다.

Intermediate Python

Introduction to Shell

Airflow와 Python을 사용하여 데이터 엔지니어링 워크플로를 구현하고 예약하는 방법을 배우세요.

Python으로 배우는 Apache Airflow 입문

데이터 엔지니어링 워크플로우를 구현하고 스케줄링하는 방법을 알아보세요.

Airflow Fundamentals

데이터 엔지니어 in Python

Data Engineer

Jinja 템플릿 활용

템플릿이란?

Jinja란?

Airflow Jinja 함수

SmtpNotifier 사용 예시

FileSensor 사용 예시

Jinja 참고 사항

연습해 봅시다!

Airflow 워크플로 유지 관리 및 모니터링 - Jinja를 활용한 템플릿 작성

Jinja를 활용한 템플릿 작성

Create Your Free Account