Airflow 핵심 개념

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

Volker Janz

Senior Developer Advocate at Astronomer

강사 소개

Volker Janz 프로필 사진

Volker Janz

Astronomer, 시니어 개발자 어드보케이트
게임 업계 데이터 엔지니어 14년 이상 경력
Airflow 버전 1.x부터 사용
Data Engineer Things 스피커, 멘토, 뉴스레터 리드

학습 내용 미리 보기

TaskFlow API로 DAG 작성
태스크 매핑과 에셋 기반 스케줄링으로 동적 워크플로 구축
재시도 및 콜백으로 오류 처리
Airflow로 SQL 워크로드 실행

각 챕터 콘텐츠 시각화

시작하기 전에

DAG, 태스크, 오퍼레이터에 익숙한 분
스케줄링 기초를 알고 있는 분

Introduction to Airflow 강의 페이지

빠른 복습

from airflow.sdk import dag, task

@dag
def star_wars_dag():


  @task
  def get_star_wars_person():
    import requests
    return requests.get("https://swapi.dev/api/people/1/").json()


  @task.bash
  def print_name(person):
    return f"echo '{person['name']}'"


  person = get_star_wars_person()
  print_name(person)

star_wars_dag()

DAG은 의존성을 가진 태스크의 집합
태스크는 개별 작업 단위
오퍼레이터 / 데코레이터는 각 태스크의 동작을 정의
의존성은 실행 순서를 정의

단순 DAG

Airflow 아키텍처

Airflow 3 아키텍처

오케스트레이션: 스케줄러, DAG 프로세서
실행: 워커, 트리거러
인터페이스 & 스토리지: API 서버, 메타데이터 DB

스케줄링 방식

# No automatic runs: trigger manually (default)
@dag(schedule=None)
def my_pipeline(): ...


# Time-based: runs every day at 6 AM
@dag(schedule="0 6 * * *")
def daily_pipeline(): ...


# Data-aware: runs when an Asset updates
@dag(schedule=[Asset("my_asset")])
def downstream_pipeline(): ...

DAG 작성 방법 두 가지

클래식 오퍼레이터

extract = PythonOperator(
    task_id="extract",
    python_callable=extract_fn)
extract >> transform

데코레이터를 사용할 수 없을 때 선택

TaskFlow API

@task
def extract():
    return {"users": 150}

data = extract()
transform(data)

간단한 Python 데코레이터
보일러플레이트 코드 감소
클래식 오퍼레이터와 혼용 가능

이 강의의 연습 문제

IDE 연습 문제 스크린샷

IDE 연습 문제: 실제 .py 파일 직접 편집
"Run this file" 클릭 또는 python3 filename.py 실행
dag.test()로 단일 프로세스에서 전체 DAG 실행

연습해 봅시다!

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