Airflow 개발 온보딩 가이드

miridih-batch-flow 레포에서 Airflow의 핵심 개념을 이해하고 dag_factory(YAML)·Python 두 방식으로 DAG를 만들거나 수정해 로컬 검증부터 PR 프리뷰·배포까지 마치는 방법.

목적

대상: Airflow를 처음 활용하는 주니어 개발자
목표: Airflow가 무엇인지 이해하고, 이 레포(miridih-batch-flow)의 구조를 읽고 첫 DAG를 만들거나 수정할 수 있게 되는 것

1. Airflow란 무엇인가

Apache Airflow는 "정해진 시간에, 정해진 순서로, 여러 작업을 자동으로 실행"해주는 오케스트레이션 도구입니다.

예를 들어 매일 새벽 5시에 다음 작업을 해야 한다고 해봅시다.

여러 서비스 DB에서 어제 하루치 데이터를 가져온다
가져온 데이터를 정제/변환한다
데이터 웨어하우스(Databricks)에 적재한다
실패하면 Slack으로 알림을 보낸다

이걸 크론(cron) + 쉘 스크립트로 짜면 "1번이 끝나야 2번이 실행되게 하기", "재시도 로직", "실패 알림", "실행 이력 확인"을 전부 직접 만들어야 합니다. Airflow는 이런 걸 표준화된 방식으로 제공합니다.

2. 핵심 개념 6가지

개념	한 줄 설명
DAG (Directed Acyclic Graph)	하나의 파이프라인 전체. "무엇을, 어떤 순서로, 언제 실행할지"의 정의. 순환하지 않는 방향 그래프라서 A→B→A 같은 무한 루프는 불가능
Task	DAG 안의 개별 작업 단위. "DB에서 추출", "Slack 알림 보내기" 같은 하나하나
Operator	Task가 "무엇을 실행하는지"의 종류. `PythonOperator`(파이썬 함수 실행), `EmptyOperator`(아무것도 안 함, 표시용), `KubernetesPodOperator`(K8s Pod 실행) 등
Scheduler	각 DAG의 스케줄(cron 표현식)을 보고 있다가 때가 되면 실행을 트리거하는 프로세스
DAG Run / Task Instance	DAG와 Task의 "정의"가 아니라 실제로 한 번 "실행된 기록". 예: 매일 실행되는 DAG는 하루마다 하나의 DAG Run이 생김
XCom	Task 간에 작은 값을 주고받는 통로. Task A의 리턴값을 Task B가 받아 쓸 수 있습니다. 6장 예제 3의 `expand_kwargs`(테이블 목록만큼 Task를 늘리는 패턴)가 이걸 사용합니다. 대용량 데이터를 XCom으로 넘기지 마세요 — 메타데이터 DB에 저장되므로 S3 경로 같은 참조만 넘깁니다

의존성은 >> 연산자로 표현합니다.

start >> extract_task >> transform_task >> load_task >> end

"start가 끝나야 extract_task 시작, extract_task가 끝나야 transform_task 시작..." 순서를 의미합니다.

Airflow UI(로컬 실행 시 http://localhost:8080)에서 이 그래프를 시각적으로 볼 수 있고, 각 Task가 성공(초록)/실패(빨강)/실행중(노랑)인지 색으로 확인할 수 있습니다.

처음에 반드시 알아야 할 두 가지: catchup과 start_date

이 두 개를 잘못 조합하면 과거 실행분이 한꺼번에 몰려서 도는 사고(backfill 폭주)가 납니다. 온보딩 중 가장 흔한 실수입니다.

start_date를 과거로 잡고 catchup: true로 두면, Airflow는 start_date부터 현재까지 밀린 DAG Run을 모두 만들어서 실행합니다. 매시간 DAG의 start_date를 6개월 전으로 두면 약 4,300개의 DAG Run이 생깁니다.

신규 DAG는 catchup: false가 기본입니다. 과거 구간을 반드시 채워야 하는 파이프라인(CDC처럼 구간 누락이 데이터 정합성 문제가 되는 경우)만 catchup: true를 쓰고, 이때는 반드시 max_active_runs: 1을 함께 지정해 동시 실행을 막습니다.

catchup: true가 필요하다고 판단되면 머지 전에 리뷰어에게 명시적으로 알려주세요.

3. 이 레포는 무엇을 하는 서비스인가

📦 https://github.com/miridih/miridih-batch-flow

miridih-batch-flow는 미리디의 중앙화된 배치(Airflow) 서비스입니다. 회사의 거의 모든 데이터 파이프라인이 여기 모여 있습니다.

CDC (Change Data Capture): 서비스 DB(MySQL, PostgreSQL 등)의 변경분을 계속 읽어서 Delta Lake로 병합
외부 광고 플랫폼 연동: Google Ads, Naver Ads, Meta Ads, Yahoo Ads 등에서 광고 성과 데이터를 가져옴
DB 스냅샷: 여러 서비스 DB 전체를 주기적으로 내보내기(export)
배치 모니터링: 위 파이프라인들이 잘 돌고 있는지 집계해서 Slack으로 요약 리포트 전송
기타 연동: Zendesk, Braze, Channeltalk, RevenueCat, SonarQube 등

기술 스택 요약:

분류	기술
오케스트레이션	Apache Airflow 2.10.5
실행 방식(Executor)	로컬: CeleryExecutor (Docker Compose) / dev·prd: Celery/KubernetesExecutor (EKS)
데이터 처리	DuckDB, delta-rs
데이터 레이크	Delta Lake (on S3)
데이터 웨어하우스	Databricks
클라우드	AWS(S3, SES, Glue, Secrets Manager), GCP(BigQuery, GCS)
알림	Slack

4. 저장소 구조 훑어보기

miridih-batch-flow/
├── dags/                         # 모든 DAG 정의가 여기 위치
│   ├── common/                   # 전체 DAG가 공유하는 유틸리티 (S3, Slack, Delta, DuckDB...)
│   ├── dag_factory/              # YAML 기반 DAG 생성 엔진 (신규 DAG는 이 방식 권장)
│   │   ├── factory_loader.py     # 진입점: YAML을 읽어 실제 DAG 객체로 변환
│   │   ├── configs/              # DAG를 정의하는 YAML 파일들 (카테고리별 폴더)
│   │   ├── dag_builder/          # pipeline_type별 DAG 생성 로직
│   │   ├── task_creator/         # task_type별 Operator 생성 로직
│   │   └── callables/            # 실제 Python 비즈니스 로직 (YAML에서 참조)
│   ├── cdc/                      # weekly snapshot 존재 (격주 덮어쓰기 DAG)
│   ├── collect_comments/         # DB 스키마/코멘트 수집 (손으로 작성한 대표 예제)
│   ├── snapshot_database/        # DB 스냅샷 내보내기
│   ├── [service]/                # 서비스별 DAG
│   │                              # (google_ads, meta_ads 등, 일부는 dag_factory로 이전됨)
│   └── operators/                # 커스텀 Operator
├── tests/                        # pytest 테스트 (dags/ 구조를 그대로 미러링)
├── docs/
│   └── miridih_dag_builder.md    # dag_factory 상세 가이드 (필독)
├── docker-compose.yaml           # 로컬 개발 환경 실행용
├── run_airflow.sh                # 로컬 Airflow 실행 스크립트
├── requirements.txt              # Python 의존성
└── CLAUDE.md                     # 이 레포의 코딩 규칙/아키텍처 원칙

가장 중요한 원칙 (CLAUDE.md): DAG는 오케스트레이션 레이어이지 비즈니스 로직 컨테이너가 아니다.

DAG 파일에는 "무엇을 어떤 순서로 실행할지"만 적고, 실제 로직(SQL 쿼리, 데이터 변환, API 호출)은 별도 함수(callables/ 또는 서비스별 모듈)에 둡니다. 이렇게 분리해야 Airflow 없이도 그 로직 함수를 pytest로 테스트할 수 있습니다.

5. DAG를 만드는 두 가지 방법

방법 A — dag_factory (YAML) · 신규 개발 시 권장

DAG를 Python 코드 대신 YAML 설정 파일로 선언합니다. 반복적인 DAG(광고 플랫폼 연동, CDC 등)를 설정값만 바꿔서 찍어낼 수 있습니다.

왜 YAML인가: 코드를 줄이려는 게 아니라, 여러 사람이 같은 레포에서 파이프라인을 함께 만들 때의 공통 규격을 두려는 것입니다. DE와 AE가 함께 YAML을 쓰고 필요하면 부품(TaskCreator)까지 함께 늘리는 구조라, 파이프라인이 무엇을 선언해야 하는지가 고정되어 있으면 서로의 PR을 읽고 리뷰할 수 있습니다.

동작 흐름:

factory_loader.py (진입점)
  → yaml_loader.py           configs/ 하위 모든 YAML을 읽음
  → DAGFactory.create_dag()  YAML의 pipeline_type을 보고 알맞은 Builder 선택
      → CommonDAGBuilder      (pipeline_type: common — 가장 많이 씀, 범용)
      → CDCMergeBuilder       (pipeline_type: cdc_merge)
      → GoogleAdsBuilder      (pipeline_type: google_ads)
      → NaverAdsBuilder       (pipeline_type: naver_ads)
  → (common의 경우) TaskCreatorFactory가 각 task의 type을 보고 Operator 생성
  → globals()[dag.dag_id] = dag   # Airflow가 이 전역 변수를 스캔해서 DAG로 인식

가장 단순한 YAML 예시 — dags/dag_factory/configs/batch_monitoring/daily_summary_report.yaml:

pipeline_type: common
dag_key: daily_batch_summary_report

dag_config:
  schedule_interval: "0 12,18 * * *"   # 매일 정오 12시, 오후 18시
  start_date: "2025-11-03"
  description: "배치 실행 결과 요약 리포트 (하루 2회: 정오, 오후 18시)"
  tags: [monitoring, summary, 6-hour]
  catchup: false
  default_args:
    retries: 2

tasks:
  - task_id: summarize_batch_execution_results
    type: python                        # PythonOperator로 생성됨
    config:
      python_callable: "dag_factory.callables.batch_monitoring.daily_summary_report.summarize_batch_execution_results"

dependencies:
  - chain: "start >> summarize_batch_execution_results >> end"

start / end는 별도로 정의하지 않아도 자동 생성되는 더미(Empty) 태스크입니다.

실제 로직은 YAML에 없고 python_callable이 가리키는 파이썬 함수 안에 있습니다 — 4장의 "DAG와 로직 분리" 원칙입니다.

YAML 저장 위치 규칙: dags/dag_factory/configs/<카테고리>/<dag_key>.yaml

YAML의 schedule_interval 키는 dag_factory 자체 스펙입니다. Airflow 2.10의 Python DAG에서는 schedule_interval 인자가 deprecated이므로 schedule=을 씁니다(방법 B 참고). YAML 키 이름이 다르다고 혼동하지 마세요.

`dependencies.chain` 문법

chain은 >> 표현식을 문자열로 그대로 적는 dag_factory 고유 DSL입니다.

dependencies:
  # 리스트로 병렬 표현 가능 (merge 후 두 삭제 태스크를 동시 실행)
  - chain: "create_temp >> export_gcs >> merge >> [delete_bq_temp, delete_gcs_files]"
  # chain을 여러 개 나열하면 각각 독립적으로 적용됨 (분기/합류 표현)
  - chain: "start >> validate_source"
  - chain: "validate_source >> [load_a, load_b] >> end"

여기 등장하는 task_id는 tasks: 블록에 정의된 것이거나 자동 생성되는 start/end여야 합니다.

지원하는 task_type은 15종 이상(python, s3_sensor, databricks_run_now, kubernetes_pod, glue_job, branch_python 등)입니다. 전체 목록은 docs/miridih_dag_builder.md 섹션 3과 CLAUDE.md를 참고하세요.

방법 B — Python DAG (직접 작성)

CDC처럼 태스크 구조가 서비스마다 정말 다르거나, 아직 마이그레이션되지 않은 기존 DAG는 Python으로 작성되어 있습니다. 신규 DAG를 순수 Python으로 작성하는 것도 금지는 아닙니다 — 반복 패턴이 없고 매우 특수한 로직이라면 Python이 더 적합할 수 있습니다. 다만 대부분의 신규 DAG는 dag_factory 쪽을 권장합니다.

판단 기준은 하나입니다 — "이런 모양의 DAG가 앞으로 몇 개가 될 것인가." 두 개 이상이 될 답이 나오면 YAML로 씁니다. 하나뿐이고 태스크 구조가 고유하며 늘어날 이유가 없다면 Python이 맞는 선택입니다.

from datetime import timedelta

with DAG(
    dag_id="merge_miricanvas_storage_docdb_cdc_to_delta",
    schedule="15 * * * *",                    # 2.10에서는 schedule_interval이 아니라 schedule
    start_date=datetime(2025, 12, 31, 0, 0, tzinfo=KST),
    tags=["internal", "cdc", "hourly", "miricanvas_storage_docdb"],
    catchup=True,                              # CDC라서 true. 신규 DAG는 false가 기본 (2장 참고)
    max_active_runs=1,                         # catchup=True와 항상 함께
    default_args={
        "on_failure_callback": send_fail_notification,
        "wait_for_downstream": True,
        "retries": 3,
        "retry_delay": timedelta(minutes=5),   # int가 아니라 timedelta로
    },
    params={...},
) as dag:
    start = EmptyOperator(task_id="start")
    fetch_config_task = PythonOperator(...)
    merge_tasks = CDCOperator.partial(...).expand_kwargs(...)
    start >> fetch_config_task >> merge_tasks >> end

6. 실제 코드로 보는 예제 3가지

예제 1 — 손으로 작성한 전형적인 Python DAG

dags/collect_comments/collect_comments_dag.py — 매주 여러 서비스 DB의 테이블/컬럼 코멘트(메타데이터 카탈로그)를 수집합니다.

with DAG(
    dag_id="ingest_intl_service_db_comments",
    schedule="0 20 * * 6",          # 매주 토요일 UTC 20:00 (KST 일요일 05:00)
    start_date=datetime(2024, 11, 15),
    catchup=False,
    tags=["internal", "metadata", "weekly"],
    default_args={"on_failure_callback": send_fail_notification},
    max_active_tasks=3,
) as dag:
    start = EmptyOperator(task_id="start")
    end = EmptyOperator(task_id="end")

    for service_name in target_services:      # 12개 서비스에 대해 반복
        task = PythonOperator(
            task_id=f"ingest_service_comments_{service_name}",
            python_callable=fetch_latest_catalog,
            op_kwargs={"service_name": service_name, "target_date": _TARGET_DATE, "env": _ENV},
        )
        start >> task >> end

주목할 점:

with DAG(...) as dag: — 파이썬 DAG를 정의하는 기본 형태
for service_name in target_services: 반복문으로 서비스마다 Task를 동적으로 생성 (fan-out 패턴)
start >> task >> end — 모든 서비스 Task가 start 뒤, end 앞에 병렬로 실행됨
max_active_tasks=3 — 12개를 동시에 다 띄우지 않고 3개씩. DB 커넥션 보호 목적
default_args={"on_failure_callback": send_fail_notification} — 실패 시 자동 Slack 알림 (dags/common/slack/)
실제 DB 조회 로직은 이 파일에 없고 fetch_latest_catalog(dags/collect_comments/service_db_catalog.py)와 SQL 템플릿(catalog_retrieval_query.py)에 분리되어 있음

예제 2 — dag_factory에서 호출되는 callable 함수

5장의 daily_summary_report.yaml이 가리키는 실제 로직 — dags/dag_factory/callables/batch_monitoring/daily_summary_report.py:

def summarize_batch_execution_results(**context):
    target_date = context.get("data_interval_end") or context.get("logical_date")
    current_hour = target_date.hour
    if current_hour == 12:
        # 전날 18시 ~ 당일 12시 구간 집계
        ...
    elif current_hour == 18:
        # 당일 12시 ~ 18시 구간 집계
        ...
    else:
        raise ValueError(f"예상하지 못한 실행 시간: {current_hour}시")

    target_dag_runs = _get_dag_runs_by_end_date(start_time, end_time)
    ...
    send_summary_to_slack(stats, sections, start_time, end_time, period_name)

Airflow REST API로 지난 시간대의 DAG 실행 이력을 가져와 성공/실패를 집계하고 Slack Block Kit 메시지로 보냅니다.

예제 3 — CDC용 YAML (실전 예시)

dags/dag_factory/configs/cdc/aicreation.yaml(요약):

pipeline_type: common
dag_key: aicreation
dag_id: merge_intl_aicreation_cdc_to_delta

dag_config:
  schedule_interval: "45 * * * *"     # 매시 45분
  start_date: "2025-04-12 23:00:00"
  deadline_hours: 1                    # 1시간 안에 안 끝나면 지연 알림
  catchup: true                        # CDC라서 구간 누락 방지 목적
  max_active_runs: 1                   # catchup=true와 항상 함께

context:                               # 이 DAG 전체에서 재사용할 변수
  db_name: aicreation
  db_type: POSTGRES
  target_schema: miridih_aicreation

tasks:
  - task_id: fetch_configuration
    type: python
    config:
      python_callable: "dag_factory.callables.cdc.fetch_configuration.fetch_cdc_configuration"

  - task_id: merge_intl_aicreation_cdc_to_delta
    type: cdc_merge
    expand_kwargs:                     # fetch_configuration 결과 개수만큼 동적으로 태스크 확장
      source_task_id: fetch_configuration

context: 블록에 정의한 값은 YAML 여러 곳에서 $context.db_name 형태로 재사용합니다.

expand_kwargs는 Airflow의 Dynamic Task Mapping으로 "테이블 개수만큼 Task를 자동으로 늘리는" 패턴입니다(테이블 20개 → 병합 Task 20개). 앞 태스크의 리턴값을 XCom으로 받아 사용합니다(2장 참고).

7. KubernetesPodOperator 활용 가이드

7.1 쿠버네티스 기본 개념

쿠버네티스(Kubernetes)를 한 문장으로 정리하면: "컨테이너(Docker 이미지)를 여러 대의 서버에 자동으로 배치하고, 죽으면 다시 살려주고, 필요하면 늘려주는 관리자"입니다.

비유로 보면 이렇습니다.

Docker 이미지 = 포장된 도시락 (실행에 필요한 코드·라이브러리·환경이 전부 들어있음)
컨테이너 = 그 도시락을 실제로 꺼내서 실행하는 것
쿠버네티스 = "지금 어느 서버(노드)에 자리가 남았는지 보고 거기다 도시락을 배달해서 실행시키고, 다 끝나면(작업 종료) 치워주는" 배달/관리 시스템
Pod = 쿠버네티스가 컨테이너를 띄우는 최소 단위 (도시락 하나가 실제로 놓이는 자리)

미리디 인프라에서는 AWS EKS(관리형 쿠버네티스)가 이 역할을 하고, dev/prd Airflow가 그 위에서 돌아갑니다(3장 기술 스택 참고).

왜 Airflow 태스크에 이게 필요한가

PythonOperator는 Airflow worker 프로세스 안에서 함수를 실행합니다. 즉 Airflow가 이미 깔아둔 파이썬 환경과 정해진 CPU/메모리를 그대로 씁니다. 그런데:

대용량 Parquet 변환처럼 무거운 작업을 그 안에서 돌리면 worker 리소스를 다 잡아먹어 다른 태스크에 영향을 줌
Airflow가 쓰는 라이브러리 버전과 이 작업에 필요한 버전이 충돌할 수 있음

KubernetesPodOperator는 "이 태스크는 Airflow worker 안에서 실행하지 말고, 완전히 독립된 Docker 이미지를 Pod로 새로 띄워서 실행해줘"라고 지시하는 오퍼레이터입니다.

Airflow Scheduler
  → "이 태스크 실행할 시간이다"
  → KubernetesPodOperator가 EKS에 요청: "이 이미지로 Pod 하나 띄워줘"
  → EKS가 여유 있는 노드를 골라 Pod 생성
  → Pod 안에서 지정한 이미지 + arguments로 프로그램 실행
  → 끝나면 결과(성공/실패)를 Airflow에 보고
  → Pod는 자동 삭제 (on_finish_action="delete_pod")

즉 Airflow는 "언제, 어떤 순서로 실행할지"만 결정하고, 실제 실행은 완전히 격리된 별도 컨테이너(Pod)에서 일어납니다. PythonOperator가 "Airflow 프로세스 안에서 직접" 실행하는 거라면, KubernetesPodOperator는 "Airflow 밖에 별도 도시락을 배달시켜서" 실행하는 것이라고 보면 됩니다.

7.2 이 레포에서의 `type: kubernetes_pod`

type: kubernetes_pod는 이 레포에서 가장 많이 쓰이는 실행 방식 중 하나입니다. Python 함수 하나로 끝나지 않고, 별도로 만든 Docker 이미지를 Kubernetes Pod로 띄워서 실행하는 태스크입니다.

⚠️ 로컬에서는 실행할 수 없습니다

10장의 ./run_airflow.sh 로컬 환경은 docker-compose.yaml 기준 CeleryExecutor로 동작하며, Pod를 스케줄할 Kubernetes 클러스터 자체가 없습니다. 실제 dev/prd는 KubernetesExecutor + EKS(miridih-de-k8s Helm 배포) 위에서 돕니다. 따라서 kubernetes_pod 태스크가 포함된 DAG를 로컬에서 트리거하면 해당 태스크는 실패하거나 무한 대기합니다. 로컬에서는 YAML 문법과 python 태스크까지만 검증하고, kubernetes_pod의 실제 동작은 11장의 PR 프리뷰 환경에서 확인하세요.

언제 python 대신 kubernetes_pod를 쓰는가

상황	선택
간단한 로직, Airflow worker의 파이썬 환경에서 바로 실행 가능	`python`
무거운 연산(대용량 Parquet 변환, BigQuery↔S3 대량 전송 등)이라 Airflow worker와 리소스를 분리하고 싶을 때	`kubernetes_pod`
Airflow의 파이썬 의존성과 충돌하는 별도 라이브러리 버전이 필요할 때	`kubernetes_pod` (독립 이미지라 의존성 격리)
여러 DAG에서 재사용할 "표준 처리 단계"(BQ 임시테이블 생성, GCS→S3 전송, Parquet→Delta 병합 등)를 이미지 하나로 통일하고 싶을 때	`kubernetes_pod`
CPU/메모리를 태스크별로 다르게 세밀히 제어하고 싶을 때	`kubernetes_pod` (`memory_request`, `cpu_request` 지정)

정리하면 "Airflow 프로세스 안에서 가볍게 끝날 일"은 python, "무겁거나 격리가 필요하거나 표준화된 배치 이미지가 이미 있는 일"은 kubernetes_pod입니다.

동작 방식 — dags/dag_factory/task_creator/kubernetes_task_creator.py

YAML의 image_name만 지정하면 ECR 이미지 경로, 환경별 태그, 노드 셀렉터, Toleration까지 KubernetesTaskCreator가 자동으로 채웁니다.

# image_name 지정 시 ECR 경로 자동 생성
if ENV_NAME == "PRODUCTION":
    image = f"{ECR_REGISTRY}/{image_name}:latest"
else:
    image = f"{ECR_REGISTRY}/{image_name}-dev:latest"   # dev/local은 -dev 태그

# 환경별 노드에만 스케줄되도록 selector/toleration 자동 설정
node_selector = {"arch": arch, "environment": env_lower}

YAML 작성자는 인프라(ECR 경로, K8s 노드 배치 규칙)를 몰라도 되고 image_name, arguments, 리소스 크기만 신경 쓰면 됩니다.

YAML 작성 예시 — 단일 태스크

- task_id: parquet_to_delta_miricanvas
  type: kubernetes_pod
  config:
    image_name: "de-airflow-operators-parquet-to-delta"   # ECR 리포지토리 이름
    name: "parquet-to-delta-loader"                        # Pod 이름 (K8s에서 보이는 이름)
    arguments:                                             # 컨테이너 실행 시 커맨드라인 인자
      - "--base-s3-path"
      - "s3://$context.landing_bucket/google_ads/$context.service_name/"
      - "--p-date"
      - "$context.p_date"
    memory_request: "4Gi"     # 기본값: 2Gi
    memory_limit: "4Gi"       # 기본값: 2Gi
    cpu_request: "2"          # 기본값: "1"
    arch: "arm64"             # 기본값: arm64
    retries: 3
    retry_delay_minutes: 5

실전 예시 — 여러 kubernetes_pod를 파이프라인으로 엮기

dags/dag_factory/configs/google_searchconsole/miricanvas.yaml은 BigQuery → GCS → S3 → Delta로 이어지는 5단계를 전부 kubernetes_pod로 구성한 예시입니다(발췌):

- task_id: export_gcs_site_impression
  type: kubernetes_pod
  config:
    image_name: "de-airflow-operators-export-bq-to-gcs"
    name: "gsc-miricanvas-export-gcs-site"
    arguments:
      - "--project-id"
      - "$context.project_id"
      - "--gcs-bucket"
      - "$context.gcs_bucket"
      - "--gcs-prefix"
      - "google_searchconsole/google_searchconsole_$context.service_name/searchdata_site_impression"
    retries: 3
    retry_delay_minutes: 5

dependencies:
  - chain: "create_temp_site_impression >> export_gcs_site_impression >> transfer_s3_site_impression >> merge_site_impression >> [delete_bq_temp_site_impression, delete_gcs_files_site_impression]"

각 단계는 전부 다른 Docker 이미지를 실행하는 별개의 Pod이고, $context.xxx로 프로젝트 ID·버킷명을 재사용합니다. 이미지 하나 = 하나의 명확한 책임(BQ 추출만, GCS 전송만...)으로 쪼개져 있어서 특정 단계만 재사용하거나 디버깅하기 쉽습니다.

주의할 점

로컬에서 동작하지 않습니다 — 위 경고 참고.
on_finish_action="delete_pod"가 기본이라 실행이 끝나면 Pod가 자동 삭제됩니다. 로그가 필요하면 Airflow UI의 Task 로그를 보세요(get_logs=True라 Pod 로그가 그대로 전달됩니다).
image vs image_name: 사내 ECR 표준 이미지는 image_name만, 외부/커스텀 전체 경로가 필요하면 image에 전체 URL을 직접 지정합니다.
리소스(memory_request/cpu_request)를 실사용량보다 너무 낮게 잡으면 OOMKilled로 실패합니다. 무거운 변환 작업이면 기존 유사 태스크의 설정을 참고하세요.

8. dags/common/ — 재사용 유틸리티 둘러보기

새 로직을 짜기 전에 항상 먼저 확인해야 하는 공유 모듈입니다.

폴더	용도
`common/aws/`	S3 조회(`s3_prefix.py`, `s3_file_exists.py`), Glue, Athena, Secrets Manager 클라이언트
`common/slack/`	Slack 알림. `slack_notification.py`(성공/실패/지연 콜백), `slack_channel.py`(DAG별 채널 라우팅)
`common/delta/`	Delta Lake 읽기/쓰기/MERGE(`upsert_delta.py`)
`common/duckdb/`	DuckDB로 MySQL/PostgreSQL에 직접 붙어서(ATTACH) 쿼리하는 헬퍼
`common/databricks/`	Databricks 카탈로그 이름 상수(`BRONZE_CATALOG`, `SILVER_CATALOG`, `GOLD_CATALOG` 등 medallion 아키텍처)
`common/db/`	DB 접속 정보(`DBCredentials`), Oracle/PostgreSQL 클라이언트
`common/gcp/`	BigQuery ↔ GCS ↔ S3 데이터 이동
`common/operator/`	커스텀 Operator
`common/utils/`	로거, 경로 관련 범용 헬퍼

예: Delta Lake에 MERGE(upsert)하려면 deltalake 라이브러리를 처음부터 다루기보다 common/delta/upsert_delta.py를 먼저 확인하세요.

9. 로컬에서 Airflow 실행해보기

# 1. 파이썬 가상환경 (테스트/린트용)
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

# 2. 로컬 Airflow 실행 (Docker Compose 기반)
./run_airflow.sh

run_airflow.sh는 AWS Secrets Manager에서 필요한 시크릿을 가져와 .env를 만들고, docker compose로 Airflow 컨테이너들을 띄웁니다. (AWS CLI 접근 권한 필요)

실행 후 http://localhost:8080 접속 → Airflow UI에서 DAG 목록/그래프/로그 확인

로컬은 CeleryExecutor, dev/prd는 별도 저장소(miridih-de-k8s)의 Kubernetes/Helm 설정으로 배포되는 Celery + KubernetesExecutor입니다. 이 차이 때문에 kubernetes_pod 태스크는 로컬에서 실행되지 않습니다(7장 참고). 실제 확인이 필요하면 10장의 PR 프리뷰 환경을 사용하세요.

10. 개발 → 배포 흐름

브랜치 및 PR 흐름

develop 에서 분기
  → feature/XXX-설명 브랜치 작업
  → 로컬 검증 (pytest + 로컬 Airflow UI에서 Import Error 없이 DAG 목록에 뜨는지 확인)
  → PR 생성 (base: develop)
  → 필요 시 preview 라벨로 PR 프리뷰 환경 확인
  → 리뷰 승인 후 develop 머지 → dev 환경 반영
  → main 머지 → 운영(prd) 반영 (Release v버전.날짜 ex. Release v10.3.20260802)

PR 프리뷰 환경 (EKS)

로컬로는 kubernetes_pod 태스크를 실행할 수 없기 때문에(7·9장), 실제 K8s 위에서 DAG를 돌려봐야 할 때는 PR 프리뷰 환경을 씁니다.

상세: https://miridih.atlassian.net/wiki/x/8gGrTg

공용 dev 서버를 여러 명이 나눠 쓰는 대신, PR 하나당 독립된 Airflow 환경을 EKS 위에 동적으로 띄우는 기능입니다. 다른 사람 작업과 뒤섞이지 않습니다.

띄우는 조건

miridih-batch-flow 레포의 PR일 것
PR이 Open 상태일 것
PR의 Labels에 preview 추가

라벨을 추가하면 ArgoCD가 감지해 8~10분 내로 전용 환경을 배포합니다.

접속

https://airflow-dev.de.miridih.com/{branch_slug}

branch_slug는 브랜치명을 URL용으로 변환한 값입니다. 예: feature/DA-001 → feature-da-001 (대문자는 소문자로, / 등 특수문자는 하이픈으로 치환)

브랜치명에는 영문 소문자·숫자·하이픈·슬래시만 사용하세요. 한글이나 그 외 특수문자가 들어가면 slug가 URL 인코딩되어 접속 URL을 예측하기 어렵습니다.

정확한 slug와 접속 URL은 ArgoCD에서 확인하는 것이 가장 확실합니다.

로그인 계정은 PR 프리뷰 문서를 참고하세요.

DAG는 어떻게 반영되는가

PR 프리뷰든 dev/prd든, DAG 코드 배포는 git pull을 수동으로 하는 게 아니라 git-sync 사이드카 컨테이너가 주기적으로 레포를 동기화하는 방식입니다. 브랜치에 커밋을 푸시하면 보통 2~3분 내(git-sync 동기화 + DAG Processor 파싱)에 Airflow UI에 반영됩니다.

Variable / Connection 추가하기

Airflow Variable이나 Connection이 필요하면 Airflow UI에 직접 입력하지 말고, AWS miridih-de 계정의 Secrets Manager에 등록하세요. 개발 환경은 애플리케이션마다 DB Pod를 새로 띄우지만, Variable/Connection은 Secrets Manager와 연동되어 있어 매번 새로 입력할 필요 없이 값이 유지됩니다.

Airflow의 Secrets Backend 설정 때문입니다 — miridih-de-k8s/airflow/production/values.yaml:

env:
  - name: "AIRFLOW__SECRETS__BACKEND"
    value: "airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend"
  - name: "AIRFLOW__SECRETS__BACKEND_KWARGS"
    value: |
      {"connections_prefix": "/de/airflow/production/connections",
       "variables_prefix": "/de/airflow/production/variables",
       "region_name": "ap-northeast-2"}

Airflow가 Variable/Connection을 조회할 때 내부 메타데이터 DB뿐 아니라 Secrets Manager의 connections_prefix/variables_prefix 경로를 직접 조회합니다. 따라서 아래 규칙에 맞는 이름으로 등록만 하면 코드나 UI 변경 없이 값을 가져다 씁니다.

# Airflow Variable → 값은 plain text로 저장
/de/airflow/{develop|production}/variables/{VARIABLE_NAME}

# Airflow Connection → 값은 key/value(JSON)로 저장
/de/airflow/{develop|production}/connections/{conn_id}

develop은 dev 환경(miridih-de-k8s/airflow/develop/, PR 프리뷰 포함), production은 운영 환경에 대응합니다. 새 외부 API 연동이나 DB 접속 정보가 필요한 DAG를 만들면 dev/prd 두 경로에 각각 등록해야 두 환경 모두에서 동작합니다.

운영(prd) 환경과의 차이

git-sync로 DAG가 자동 배포되는 점은 dev와 같고, 운영은 PR별로 뜨는 게 아니라 develop/main 브랜치를 추적하는 고정 환경이라는 점만 다릅니다.

11. 테스트는 어떻게 작성하는가

이 레포의 원칙: 모든 코드는 pytest로 테스트하고, Airflow 인프라 없이 실행 가능해야 합니다.

tests/ 폴더는 dags/ 구조를 그대로 미러링합니다.

11.1 로직 함수 단위 테스트

예: tests/batch_monitoring/test_daily_summary_report_dag.py

from dag_factory.callables.batch_monitoring import daily_summary_report as m

class TestTimeRange:
    def test_morning_report_range(self):
        """12시 리포트는 전날 18시 ~ 당일 12시를 커버해야 한다"""
        execution_date = pendulum.datetime(2025, 11, 4, 12, 0, tz="Asia/Seoul")
        with patch.object(m, "_get_dag_runs_by_end_date", return_value=[]) as mock_runs, ...:
            m.summarize_batch_execution_results(execution_date=execution_date)
            call_args = mock_runs.call_args[0]
        assert call_args[0] == pendulum.datetime(2025, 11, 3, 18, 0, 0, tz="Asia/Seoul")

핵심 패턴:

실제 프로젝트 코드를 import해서 테스트
S3/Slack/DB 같은 외부 의존성은 mock으로 교체
patch.object로 내부 호출을 감시하며 Given/When/Then 스타일로 검증

11.2 DAG 로딩(import) 테스트 — dag_factory에서 특히 중요

dag_factory 구조에서는 YAML 하나의 오타가 파싱 단계에서 터집니다. 로직 테스트를 다 통과해도 DAG가 아예 뜨지 않을 수 있으므로, "모든 YAML이 정상적으로 DAG 객체로 만들어지는가"를 검증하는 테스트가 가장 값싸고 효과적인 안전망입니다.

def test_no_import_errors():
    """configs/ 하위 모든 YAML이 DAG로 로딩되어야 한다"""
    dagbag = DagBag(dag_folder="dags/", include_examples=False)
    assert not dagbag.import_errors, dagbag.import_errors

def test_dag_ids_are_unique():
    """dag_id 중복은 조용히 DAG를 덮어쓰므로 반드시 검증"""
    ...

11.3 실행 방법

pytest tests/ -v                                 # 전체 실행
pytest tests/batch_monitoring/ -v                # 특정 폴더만
pytest tests/ -v --cov=dags --cov-report=html    # 커버리지 포함
flake8 dags/ tests/                              # 스타일 검사

12. 첫 작업을 위한 체크리스트

새 DAG를 추가하거나 기존 DAG를 수정할 때:

13. 참고문서

[dag builder] Analytics engineer를 위한 Airflow 데이터 플랫폼 — 플랫폼 지향점 원문 (사내 Confluence, 개인 스페이스)
[dag builder] DAG Builder / TaskCreator — 문제 정의와 추진 로드맵 원문 (사내 Confluence, 개인 스페이스)
https://miridih.atlassian.net/wiki/x/8gGrTg — PR 프리뷰, Variable/Connection 등록 규칙, DAG 자동 배포(git-sync) 상세
docs/miridih_dag_builder.md (레포 내 파일) — 지원 task_type 전체 목록과 필드 (필독)
CLAUDE.md (레포 내 파일) — 레포 코딩 규칙/아키텍처 원칙
Airflow 공식 문서: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/index.html