Hands-on: Mini Data Pipeline

mkdir -p ~/airflow-study && cd ~/airflow-study
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'
mkdir -p ./dags ./logs ./plugins ./config ./data
echo "AIRFLOW_UID=$(id -u)" > .env

docker compose up airflow-init   # 메타데이터 DB 초기화 (최초 1회)
docker compose up -d             # 전체 스택 기동 (webserver/scheduler/worker...)

# x-airflow-common 의 volumes: 아래에 추가
    - ${AIRFLOW_PROJ_DIR:-.}/data:/opt/airflow/data

import csv
import json
import sqlite3
from pathlib import Path

import pendulum

# Airflow 3.x / 2.x 둘 다 대응
try:
    from airflow.sdk import dag, task, get_current_context
except ImportError:
    from airflow.decorators import dag, task
    from airflow.operators.python import get_current_context

BASE_DIR = Path("/opt/airflow/data")


@dag(
    dag_id="mini_data_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2026, 7, 1, tz="Asia/Seoul"),
    catchup=False,
    tags=["study", "s3", "bq", "dt-platform"],
)
def mini_data_pipeline():
    @task
    def extract_to_s3_like() -> str:
        """S3에 raw event를 쌓는 상황을 로컬 폴더로 흉내. (실무: s3://bucket/.../dt=.../)"""
        ds = get_current_context()["ds"]
        raw_dir = BASE_DIR / "s3_like" / "raw" / "event_bus" / f"dt={ds}"
        raw_dir.mkdir(parents=True, exist_ok=True)
        raw_path = raw_dir / "events.jsonl"

        events = [
            {"event_id": "e1", "user_id": 101, "event_name": "view_item", "amount": 0},
            {"event_id": "e2", "user_id": 102, "event_name": "purchase", "amount": 13000},
            {"event_id": "e3", "user_id": 101, "event_name": "purchase", "amount": 7000},
        ]
        with raw_path.open("w") as f:
            for event in events:
                f.write(json.dumps(event) + "\n")
        return str(raw_path)

    @task
    def validate(raw_path: str) -> str:
        """데이터 품질 체크: 필수 컬럼 누락 / row 0개면 실패."""
        required_keys = {"event_id", "user_id", "event_name", "amount"}
        count = 0
        with open(raw_path) as f:
            for line in f:
                row = json.loads(line)
                missing = required_keys - row.keys()
                if missing:
                    raise ValueError(f"Missing keys: {missing}")
                count += 1
        if count == 0:
            raise ValueError("No data found")
        return raw_path

    @task
    def transform(validated_raw_path: str) -> str:
        """raw JSONL → curated CSV. (실무: Spark job / dbt model / SQL transform)"""
        ds = get_current_context()["ds"]
        curated_dir = BASE_DIR / "s3_like" / "curated" / "event_bus" / f"dt={ds}"
        curated_dir.mkdir(parents=True, exist_ok=True)
        csv_path = curated_dir / "events.csv"

        with open(validated_raw_path) as infile, csv_path.open("w", newline="") as outfile:
            writer = csv.DictWriter(
                outfile,
                fieldnames=["dt", "event_id", "user_id", "event_name", "amount"],
            )
            writer.writeheader()
            for line in infile:
                row = json.loads(line)
                writer.writerow({"dt": ds, **{k: row[k] for k in
                    ["event_id", "user_id", "event_name", "amount"]}})
        return str(csv_path)

    @task
    def load_to_bq_like(csv_path: str) -> str:
        """BigQuery 적재를 SQLite insert로 흉내. (실무: BigQuery load job / connector)"""
        warehouse_dir = BASE_DIR / "warehouse"
        warehouse_dir.mkdir(parents=True, exist_ok=True)
        db_path = warehouse_dir / "bq_like.db"

        conn = sqlite3.connect(db_path)
        cur = conn.cursor()
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS fact_event (
                dt TEXT, event_id TEXT PRIMARY KEY, user_id INTEGER,
                event_name TEXT, amount INTEGER
            )
            """
        )
        with open(csv_path) as f:
            for row in csv.DictReader(f):
                cur.execute(
                    """INSERT OR REPLACE INTO fact_event
                       (dt, event_id, user_id, event_name, amount)
                       VALUES (?, ?, ?, ?, ?)""",
                    (row["dt"], row["event_id"], int(row["user_id"]),
                     row["event_name"], int(row["amount"])),
                )
        conn.commit()
        conn.close()
        return str(db_path)

    @task
    def data_quality_check(db_path: str) -> None:
        """적재 결과 검증. (실무: row count / null / 중복 / partition check)"""
        conn = sqlite3.connect(db_path)
        cur = conn.cursor()
        cur.execute("SELECT COUNT(*) FROM fact_event")
        row_count = cur.fetchone()[0]
        cur.execute("SELECT SUM(amount) FROM fact_event")
        total_amount = cur.fetchone()[0]
        conn.close()
        if row_count == 0:
            raise ValueError("Loaded table is empty")
        print(f"row_count={row_count}")
        print(f"total_amount={total_amount}")

    # 반환값을 다음 task 인자로 넘기는 것 = 의존성 정의
    raw_path = extract_to_s3_like()
    validated_path = validate(raw_path)
    csv_path = transform(validated_path)
    db_path = load_to_bq_like(csv_path)
    data_quality_check(db_path)


mini_data_pipeline()

helm repo add apache-airflow https://airflow.apache.org
helm repo update
# Chart가 실제로 어떤 K8s YAML로 변환되는지 확인 (클러스터 없이도 가능)
helm template airflow apache-airflow/airflow --namespace airflow > rendered-airflow.yaml