데이터 엔지니어링 파이프라인 완전 가이드 2025: ETL/ELT, Spark, Airflow, 실시간 스트리밍

데이터 엔지니어의 핵심 역할
├── 데이터 수집 (Ingestion)
│   ├── API, DB, 파일, 스트림에서 데이터 추출
│   └── 다양한 소스를 통합
├── 데이터 변환 (Transformation)
│   ├── 정제, 정규화, 집계
│   └── 비즈니스 로직 적용
├── 데이터 저장 (Storage)
│   ├── 데이터 웨어하우스 설계
│   └── 데이터 레이크 구축
├── 파이프라인 오케스트레이션
│   ├── 워크플로 자동화
│   └── 스케줄링 및 모니터링
└── 데이터 품질 관리
    ├── 데이터 검증
    └── 데이터 계약 (Data Contract)

프로그래밍       : Python, SQL, Scala/Java
데이터 처리      : Spark, Flink, Beam
오케스트레이션   : Airflow, Dagster, Prefect
스트리밍         : Kafka, Kinesis, Pub/Sub
변환 도구        : dbt, Dataform
클라우드          : AWS(Redshift, Glue), GCP(BigQuery), Azure(Synapse)
컨테이너         : Docker, Kubernetes
IaC              : Terraform, Pulumi
모니터링         : Datadog, Grafana, Monte Carlo

ETL 흐름:
소스 DB ──Extract──→ 변환 서버 ──Transform──→ Load──→ 데이터 웨어하우스
                      (ETL 서버)

# 전통적 ETL 예시 (Python)
import pandas as pd
from sqlalchemy import create_engine

# Extract: 소스 DB에서 데이터 추출
source_engine = create_engine('postgresql://source_db:5432/sales')
raw_data = pd.read_sql('SELECT * FROM orders WHERE date >= %s', source_engine, params=['2025-01-01'])

# Transform: 데이터 변환
transformed = raw_data.copy()
transformed['total_with_tax'] = transformed['total'] * 1.1
transformed['order_month'] = pd.to_datetime(transformed['order_date']).dt.to_period('M')
transformed = transformed.dropna(subset=['customer_id'])
transformed = transformed[transformed['total'] > 0]

# Load: 웨어하우스에 적재
wh_engine = create_engine('postgresql://warehouse:5432/analytics')
transformed.to_sql('fact_orders', wh_engine, if_exists='append', index=False)

ELT 흐름:
소스 DB ──Extract──→ Load──→ 데이터 웨어하우스 ──Transform──→ 분석용 테이블
                                 (dbt 등으로 변환)

-- ELT: 웨어하우스 내에서 dbt로 변환
-- models/marts/fact_orders.sql
WITH source_orders AS (
    SELECT * FROM raw.orders
    WHERE order_date >= '2025-01-01'
),
cleaned AS (
    SELECT
        order_id,
        customer_id,
        total,
        total * 1.1 AS total_with_tax,
        DATE_TRUNC('month', order_date) AS order_month,
        order_date
    FROM source_orders
    WHERE customer_id IS NOT NULL
      AND total > 0
)
SELECT * FROM cleaned

배치 처리:
[데이터 축적] ──→ [일괄 처리] ──→ [결과 저장]
  (1시간/1일)      (Spark 등)      (웨어하우스)

특징:
- 높은 처리량 (Throughput)
- 지연 시간이 김 (분~시간)
- 비용 효율적
- 재처리 용이

스트림 처리:
[이벤트 발생] ──→ [즉시 처리] ──→ [실시간 결과]
  (연속적)        (Flink 등)      (대시보드/알림)

특징:
- 낮은 지연 시간 (밀리초~초)
- 연속 처리
- 복잡한 장애 처리
- 이벤트 순서 관리 필요

배치를 선택하는 경우:
  - 일/주/월 단위 보고서
  - 대규모 데이터 집계
  - 비용 최적화 우선
  - 실시간 필요 없음

스트림을 선택하는 경우:
  - 실시간 대시보드
  - 사기 탐지
  - 실시간 추천
  - IoT 센서 데이터
  - 알림/알럿

하이브리드 (Lambda/Kappa):
  - 배치 + 스트림 동시 운영
  - 실시간 근사치 + 배치 정확 결과

Spark 아키텍처:
┌─────────────────────────────────┐
│         Spark Application       │
├─────────────────────────────────┤
│  SparkSQL  │ Streaming │  MLlib │
├─────────────────────────────────┤
│        DataFrame / Dataset      │
├─────────────────────────────────┤
│         RDD (Core Engine)       │
├─────────────────────────────────┤
│  Standalone │ YARN │ Mesos │ K8s│
└─────────────────────────────────┘

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, avg, count, when, lit

# SparkSession 생성
spark = SparkSession.builder \
    .appName("DataPipeline") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

# 데이터 읽기
orders_df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("s3://data-lake/raw/orders/")

users_df = spark.read.parquet("s3://data-lake/raw/users/")

# 데이터 변환
result = orders_df \
    .filter(col("status") == "completed") \
    .join(users_df, orders_df.user_id == users_df.id, "inner") \
    .groupBy("user_id", "username") \
    .agg(
        count("order_id").alias("total_orders"),
        sum("amount").alias("total_spent"),
        avg("amount").alias("avg_order_value")
    ) \
    .filter(col("total_orders") > 5)

# 결과 저장
result.write \
    .mode("overwrite") \
    .partitionBy("order_month") \
    .parquet("s3://data-lake/processed/user_summary/")

# 임시 뷰 등록
orders_df.createOrReplaceTempView("orders")
users_df.createOrReplaceTempView("users")

# SQL로 분석
monthly_revenue = spark.sql("""
    SELECT
        DATE_TRUNC('month', order_date) AS month,
        COUNT(DISTINCT user_id) AS unique_customers,
        COUNT(*) AS total_orders,
        SUM(amount) AS revenue,
        AVG(amount) AS avg_order_value
    FROM orders
    WHERE status = 'completed'
    GROUP BY DATE_TRUNC('month', order_date)
    ORDER BY month
""")

monthly_revenue.show()

# 파티셔닝 최적화
# 파티션 수 확인
print(f"Partitions: {orders_df.rdd.getNumPartitions()}")

# 재파티셔닝 (셔플 발생)
orders_repartitioned = orders_df.repartition(100, "order_date")

# 합병 (셔플 없이 파티션 수 줄이기)
orders_coalesced = orders_df.coalesce(50)

# 캐싱
from pyspark.storagelevel import StorageLevel

# 메모리 캐싱
orders_df.cache()
orders_df.count()  # 캐시 트리거

# 메모리 + 디스크 캐싱
users_df.persist(StorageLevel.MEMORY_AND_DISK)

# 캐시 해제
orders_df.unpersist()

# 브로드캐스트 조인 (작은 테이블)
from pyspark.sql.functions import broadcast

# 작은 테이블을 브로드캐스트
result = orders_df.join(
    broadcast(users_df),
    orders_df.user_id == users_df.id
)

# AQE (Adaptive Query Execution) 활성화
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

Spark 성능 최적화 체크리스트:
  [ ] 적절한 파티션 수 설정 (코어 수의 2~3배)
  [ ] 데이터 스큐 처리 (salting, AQE)
  [ ] 브로드캐스트 조인 활용
  [ ] 불필요한 셔플 최소화
  [ ] 컬럼 프루닝 (필요한 컬럼만 선택)
  [ ] Predicate Pushdown 활용
  [ ] 캐싱 전략 수립
  [ ] 직렬화 포맷 최적화 (Parquet, ORC)

Airflow 아키텍처:
┌──────────────────────────────────┐
│           Web Server             │
│    (UI / REST API)               │
├──────────────────────────────────┤
│           Scheduler              │
│    (DAG 파싱, 태스크 스케줄링)    │
├──────────────────────────────────┤
│           Executor               │
│  (Local/Celery/Kubernetes)       │
├──────────────────────────────────┤
│        Metadata Database         │
│    (PostgreSQL/MySQL)            │
└──────────────────────────────────┘

# dags/etl_pipeline.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['data-team@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=2),
}

with DAG(
    dag_id='daily_etl_pipeline',
    default_args=default_args,
    description='일별 ETL 파이프라인',
    schedule_interval='0 6 * * *',  # 매일 오전 6시
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=['etl', 'daily'],
    max_active_runs=1,
) as dag:

    # 센서: 파일 도착 대기
    wait_for_data = FileSensor(
        task_id='wait_for_data',
        filepath='/data/raw/daily_export.csv',
        poke_interval=300,  # 5분마다 확인
        timeout=3600,       # 최대 1시간 대기
        mode='poke',
    )

    # 추출
    def extract_data(**context):
        import pandas as pd
        execution_date = context['ds']
        df = pd.read_csv(f'/data/raw/daily_export_{execution_date}.csv')
        df.to_parquet(f'/data/staging/extract_{execution_date}.parquet')
        return len(df)

    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_data,
    )

    # 변환
    def transform_data(**context):
        import pandas as pd
        execution_date = context['ds']
        df = pd.read_parquet(f'/data/staging/extract_{execution_date}.parquet')

        # 데이터 정제
        df = df.dropna(subset=['customer_id', 'amount'])
        df = df[df['amount'] > 0]
        df['amount_with_tax'] = df['amount'] * 1.1

        df.to_parquet(f'/data/staging/transform_{execution_date}.parquet')
        return len(df)

    transform = PythonOperator(
        task_id='transform',
        python_callable=transform_data,
    )

    # 적재
    load = PostgresOperator(
        task_id='load',
        postgres_conn_id='warehouse',
        sql='sql/load_daily_orders.sql',
    )

    # 데이터 품질 검증
    def validate_data(**context):
        execution_date = context['ds']
        # 레코드 수 검증, NULL 체크, 범위 검증
        row_count = context['ti'].xcom_pull(task_ids='transform')
        if row_count < 100:
            raise ValueError(f'레코드 수 부족: {row_count}')

    validate = PythonOperator(
        task_id='validate',
        python_callable=validate_data,
    )

    # 알림
    notify = BashOperator(
        task_id='notify',
        bash_command='echo "ETL 완료: {{ ds }}" | mail -s "ETL Success" team@company.com',
    )

    # 의존 관계 정의
    wait_for_data >> extract >> transform >> load >> validate >> notify

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=['etl'],
)
def modern_etl_pipeline():

    @task()
    def extract():
        """소스에서 데이터 추출"""
        import pandas as pd
        df = pd.read_csv('/data/raw/orders.csv')
        return df.to_dict()

    @task()
    def transform(raw_data: dict):
        """데이터 변환 및 정제"""
        import pandas as pd
        df = pd.DataFrame(raw_data)
        df = df[df['amount'] > 0]
        df['processed_at'] = datetime.now().isoformat()
        return df.to_dict()

    @task()
    def load(transformed_data: dict):
        """웨어하우스에 적재"""
        import pandas as pd
        df = pd.DataFrame(transformed_data)
        # 웨어하우스에 적재 로직
        print(f"Loaded {len(df)} records")

    # 자동 의존 관계 설정
    raw = extract()
    transformed = transform(raw)
    load(transformed)

# DAG 인스턴스화
modern_etl_pipeline()

# Connections: UI 또는 CLI로 설정
# airflow connections add 'warehouse' \
#   --conn-type 'postgres' \
#   --conn-host 'warehouse.example.com' \
#   --conn-port 5432 \
#   --conn-login 'etl_user' \
#   --conn-password 'secret'

# Variables
from airflow.models import Variable

env = Variable.get("environment", default_var="dev")
config = Variable.get("pipeline_config", deserialize_json=True)

# XCom: 태스크 간 데이터 전달
def producer_task(**context):
    context['ti'].xcom_push(key='row_count', value=1000)

def consumer_task(**context):
    row_count = context['ti'].xcom_pull(
        task_ids='producer',
        key='row_count'
    )
    print(f"이전 태스크 결과: {row_count}행")

# dags/dynamic_dag_factory.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# 설정 기반 동적 DAG 생성
configs = [
    {"name": "sales", "source": "mysql", "schedule": "@daily"},
    {"name": "users", "source": "postgres", "schedule": "@hourly"},
    {"name": "logs", "source": "s3", "schedule": "0 */6 * * *"},
]

def create_etl_dag(config):
    dag = DAG(
        dag_id=f"etl_{config['name']}",
        schedule_interval=config['schedule'],
        start_date=datetime(2025, 1, 1),
        catchup=False,
    )

    def process(**kwargs):
        print(f"Processing {config['name']} from {config['source']}")

    with dag:
        PythonOperator(
            task_id='process',
            python_callable=process,
        )

    return dag

for config in configs:
    globals()[f"etl_{config['name']}"] = create_etl_dag(config)

Kafka 아키텍처:
Producer ──→ Broker(Topic/Partition) ──→ Consumer Group
                  │
                  ├── Partition 0: [msg1, msg2, msg3...]
                  ├── Partition 1: [msg4, msg5, msg6...]
                  └── Partition 2: [msg7, msg8, msg9...]

# Kafka Producer (Python)
from confluent_kafka import Producer
import json

config = {
    'bootstrap.servers': 'kafka-broker:9092',
    'acks': 'all',
    'retries': 3,
    'linger.ms': 10,
    'batch.size': 16384,
}

producer = Producer(config)

def delivery_callback(err, msg):
    if err:
        print(f'전송 실패: {err}')
    else:
        print(f'전송 완료: {msg.topic()} [{msg.partition()}]')

# 메시지 전송
for i in range(100):
    event = {
        'user_id': f'user_{i}',
        'action': 'page_view',
        'timestamp': '2025-03-25T10:00:00Z',
        'page': '/products'
    }
    producer.produce(
        topic='user-events',
        key=str(event['user_id']),
        value=json.dumps(event),
        callback=delivery_callback
    )

producer.flush()

# Kafka Consumer (Python)
from confluent_kafka import Consumer
import json

config = {
    'bootstrap.servers': 'kafka-broker:9092',
    'group.id': 'analytics-consumer',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,
}

consumer = Consumer(config)
consumer.subscribe(['user-events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        event = json.loads(msg.value().decode('utf-8'))
        print(f"수신: {event['user_id']} - {event['action']}")

        # 수동 커밋
        consumer.commit(asynchronous=False)
finally:
    consumer.close()

# PyFlink 스트림 처리
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
env.enable_checkpointing(60000)  # 60초마다 체크포인트

t_env = StreamTableEnvironment.create(env)

# Kafka 소스 테이블 정의
t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id STRING,
        action STRING,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'properties.group.id' = 'flink-processor',
        'format' = 'json',
        'scan.startup.mode' = 'latest-offset'
    )
""")

# 윈도우 집계
t_env.execute_sql("""
    CREATE TABLE page_view_stats (
        window_start TIMESTAMP(3),
        window_end TIMESTAMP(3),
        page STRING,
        view_count BIGINT,
        unique_users BIGINT
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:postgresql://analytics-db:5432/stats',
        'table-name' = 'page_view_stats',
        'driver' = 'org.postgresql.Driver'
    )
""")

t_env.execute_sql("""
    INSERT INTO page_view_stats
    SELECT
        window_start,
        window_end,
        action AS page,
        COUNT(*) AS view_count,
        COUNT(DISTINCT user_id) AS unique_users
    FROM TABLE(
        TUMBLE(TABLE user_events, DESCRIPTOR(event_time), INTERVAL '5' MINUTE)
    )
    GROUP BY window_start, window_end, action
""")

전달 보장 수준:
  At-most-once  : 메시지 유실 가능, 중복 없음
  At-least-once : 메시지 유실 없음, 중복 가능
  Exactly-once  : 메시지 유실 없음, 중복 없음 (가장 어려움)

Kafka + Flink Exactly-Once:
  1. Kafka 트랜잭셔널 프로듀서
  2. Flink 체크포인트 (Chandy-Lamport)
  3. Two-Phase Commit Protocol
  4. Kafka Consumer 오프셋을 체크포인트와 연동

dbt 프로젝트 구조:
my_dbt_project/
├── dbt_project.yml
├── profiles.yml
├── models/
│   ├── staging/
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── _staging_sources.yml
│   ├── intermediate/
│   │   └── int_order_items_grouped.sql
│   └── marts/
│       ├── dim_customers.sql
│       ├── fact_orders.sql
│       └── _marts_schema.yml
├── tests/
│   └── assert_positive_revenue.sql
├── macros/
│   └── generate_schema_name.sql
└── seeds/
    └── country_codes.csv

-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),
renamed AS (
    SELECT
        id AS order_id,
        user_id AS customer_id,
        amount AS order_amount,
        status AS order_status,
        created_at AS ordered_at
    FROM source
    WHERE status != 'cancelled'
)
SELECT * FROM renamed

-- models/marts/fact_orders.sql
{{ config(materialized='incremental', unique_key='order_id') }}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),
customers AS (
    SELECT * FROM {{ ref('dim_customers') }}
)
SELECT
    o.order_id,
    o.customer_id,
    c.customer_name,
    c.customer_segment,
    o.order_amount,
    o.order_amount * 1.1 AS amount_with_tax,
    o.order_status,
    o.ordered_at
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id

{% if is_incremental() %}
WHERE o.ordered_at > (SELECT MAX(ordered_at) FROM {{ this }})
{% endif %}

# models/staging/_staging_sources.yml
version: 2

sources:
  - name: raw
    database: raw_db
    schema: public
    tables:
      - name: orders
        loaded_at_field: _loaded_at
        freshness:
          warn_after:
            count: 12
            period: hour
          error_after:
            count: 24
            period: hour
        columns:
          - name: id
            tests:
              - unique
              - not_null
          - name: amount
            tests:
              - not_null

  - name: raw
    tables:
      - name: customers
        columns:
          - name: id
            tests:
              - unique
              - not_null

# models/marts/_marts_schema.yml
version: 2

models:
  - name: fact_orders
    description: "주문 팩트 테이블"
    columns:
      - name: order_id
        description: "주문 고유 ID"
        tests:
          - unique
          - not_null
      - name: order_amount
        tests:
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id

-- tests/assert_positive_revenue.sql
-- 커스텀 테스트: 매출이 양수인지 검증
SELECT order_id, order_amount
FROM {{ ref('fact_orders') }}
WHERE order_amount < 0

# 모든 모델 빌드
dbt run

# 특정 모델만 빌드
dbt run --select fact_orders

# 모델 + 다운스트림 빌드
dbt run --select stg_orders+

# 테스트 실행
dbt test

# 문서 생성
dbt docs generate
dbt docs serve

# 소스 신선도 확인
dbt source freshness

# 시드 데이터 로드
dbt seed

# 전체 파이프라인 (빌드 + 테스트)
dbt build

-- BigQuery: 파티션 + 클러스터링
CREATE TABLE analytics.fact_orders
PARTITION BY DATE(ordered_at)
CLUSTER BY customer_segment, order_status
AS
SELECT
    order_id,
    customer_id,
    customer_segment,
    order_amount,
    order_status,
    ordered_at
FROM staging.orders;

-- 비용 예측 (dry run)
-- 1TB 스캔 = 약 $5 (온디맨드)

-- Snowflake: 웨어하우스 생성 및 관리
CREATE WAREHOUSE etl_wh
    WITH WAREHOUSE_SIZE = 'MEDIUM'
    AUTO_SUSPEND = 300
    AUTO_RESUME = TRUE
    MIN_CLUSTER_COUNT = 1
    MAX_CLUSTER_COUNT = 3;

-- 데이터 로드
COPY INTO raw.orders
FROM @my_s3_stage/orders/
FILE_FORMAT = (TYPE = 'PARQUET')
PATTERN = '.*[.]parquet';

전통적 데이터 레이크 문제점:
  - ACID 트랜잭션 없음
  - 스키마 강제 없음
  - 시간 여행(Time Travel) 불가
  - 작은 파일 문제

레이크하우스 테이블 포맷이 해결:
  Delta Lake  : Databricks 주도, Spark 통합 최강
  Apache Iceberg : Netflix 개발, 벤더 중립
  Apache Hudi : Uber 개발, 증분 처리 특화

# Delta Lake 예시 (PySpark)
from delta.tables import DeltaTable

# Delta 테이블 생성
orders_df.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .save("s3://data-lake/delta/orders")

# UPSERT (Merge)
delta_table = DeltaTable.forPath(spark, "s3://data-lake/delta/orders")

delta_table.alias("target").merge(
    new_orders_df.alias("source"),
    "target.order_id = source.order_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# 시간 여행
old_data = spark.read \
    .format("delta") \
    .option("versionAsOf", 5) \
    .load("s3://data-lake/delta/orders")

import great_expectations as gx

context = gx.get_context()

# 데이터 소스 연결
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_csv_asset("orders", filepath_or_buffer="orders.csv")

# Expectation Suite 정의
suite = context.add_expectation_suite("orders_validation")

# 기대치 정의
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=100000
    )
)

# 검증 실행
results = context.run_checkpoint(
    checkpoint_name="orders_checkpoint"
)
print(f"성공: {results.success}")

# data-contracts/orders-contract.yaml
dataContractSpecification: 0.9.3
id: orders-contract
info:
  title: Orders Data Contract
  version: 1.0.0
  owner: data-team
  contact:
    email: data-team@company.com

schema:
  type: object
  properties:
    order_id:
      type: string
      description: "고유 주문 ID"
      required: true
      unique: true
    customer_id:
      type: string
      required: true
    amount:
      type: number
      minimum: 0
      maximum: 100000
    status:
      type: string
      enum: ["pending", "completed", "cancelled"]
    created_at:
      type: timestamp
      required: true

quality:
  completeness:
    - field: order_id
      threshold: 100
    - field: customer_id
      threshold: 99.9
  freshness:
    maxDelay: "PT1H"  # 1시간 이내

선택 가이드:
├── 대규모 기업, 복잡한 워크플로 → Airflow
├── 데이터 자산 중심 사고 → Dagster
├── 빠른 시작, Python 네이티브 → Prefect
└── 노코드/로우코드 선호 → Mage

모던 데이터 스택 (2025):

데이터 소스                수집/통합            저장              변환            분석/BI
──────────           ──────────         ──────────       ──────────      ──────────
 SaaS APIs  ──┐
 Databases  ──┼──→ Fivetran/Airbyte ──→ Snowflake  ──→   dbt      ──→  Looker
 Event Logs ──┤                         BigQuery       Dataform       Metabase
 Files      ──┘                         Redshift                     Tableau

 Kafka/     ──────→ Flink/Spark   ──→ Delta Lake  ──→  Spark SQL  ──→ 실시간
 Kinesis             Streaming         Iceberg                       대시보드

                                 오케스트레이션: Airflow / Dagster
                                 품질: Great Expectations / dbt tests
                                 카탈로그: DataHub / Atlan / OpenMetadata
                                 모니터링: Monte Carlo / Datadog