데이터 파이프라인: ETL + 스트리밍

[Batch ETL]
- 매일/시간 단위 처리
- 정확한 결과
- 도구: Airflow, dbt, Spark

[Stream Processing]
- 실시간 처리
- 1초~분 단위 응답
- 도구: Kafka Streams, Flink

[CDC (Change Data Capture)]
- DB 변경 → 다른 시스템 동기화
- Debezium, AWS DMS

[Source]
- PostgreSQL (운영 DB)
- API (외부 데이터)
- Files (S3)
        ↓
[Extract]
- Airflow operator로 추출
- 시간/일 단위 스케줄
        ↓
[Transform]
- dbt 또는 Python으로 변환
- 정규화·집계
        ↓
[Load]
- BigQuery / Snowflake
- 분석용 DW
        ↓
[BI]
- Looker / Metabase / Tableau

-- models/staging/orders.sql
WITH source AS (
  SELECT * FROM {{ source('app_db', 'orders') }}
)
SELECT
  id as order_id,
  user_id,
  total,
  status,
  created_at,
  paid_at,
  EXTRACT(EPOCH FROM (paid_at - created_at)) as time_to_pay_seconds
FROM source
WHERE deleted_at IS NULL


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

SELECT
  DATE(created_at) as date,
  COUNT(DISTINCT user_id) as unique_customers,
  COUNT(*) as order_count,
  SUM(total) as total_revenue,
  AVG(total) as avg_order_value
FROM {{ ref('orders') }}
WHERE status = 'completed'
{% if is_incremental() %}
  AND created_at > (SELECT MAX(date) FROM {{ this }})
{% endif %}
GROUP BY 1

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    'daily_revenue',
    schedule='0 1 * * *',  # 매일 새벽 1시
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

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

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

    load = PythonOperator(
        task_id='load_bigquery',
        python_callable=load_to_bq,
    )

    notify = PythonOperator(
        task_id='notify_slack',
        python_callable=send_slack_summary,
    )

    extract >> transform >> load >> notify

// PostgreSQL → Kafka via Debezium
// 실시간 동기화


// 사용 사례
// - PostgreSQL → Elasticsearch (검색 인덱스)
// - PostgreSQL → BigQuery (분석)
// - PostgreSQL → Redis (캐시 갱신)


// PostgreSQL Logical Replication
ALTER TABLE products REPLICA IDENTITY FULL;
SELECT pg_create_logical_replication_slot('cdc_slot', 'pgoutput');


// Debezium connector
{
  "name": "products-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "myapp",
    "schema.include.list": "public",
    "table.include.list": "public.products",
    "topic.prefix": "myapp"
  }
}


// → products 변경 시 Kafka 토픽 myapp.public.products로 자동 발행
// → 다른 시스템이 구독해서 동기화

// Kafka Streams (간단한 예)
import { Kafka } from 'kafkajs';

const kafka = new Kafka({ brokers: ['kafka:9092'] });
const consumer = kafka.consumer({ groupId: 'analytics' });
const producer = kafka.producer();


// 실시간 매출 집계 (1분 단위)
const windowAggregates = new Map<string, { count: number; total: number }>();

await consumer.subscribe({ topic: 'orders' });
await consumer.run({
  eachMessage: async ({ message }) => {
    const order = JSON.parse(message.value!.toString());
    const minute = Math.floor(Date.now() / 60000);
    const key = `${minute}`;

    const agg = windowAggregates.get(key) ?? { count: 0, total: 0 };
    agg.count++;
    agg.total += order.total;
    windowAggregates.set(key, agg);
  },
});


// 매분 집계 결과 발행
setInterval(async () => {
  const previousMinute = Math.floor(Date.now() / 60000) - 1;
  const agg = windowAggregates.get(`${previousMinute}`);
  if (agg) {
    await producer.send({
      topic: 'metrics.revenue.minutely',
      messages: [{ value: JSON.stringify({ minute: previousMinute, ...agg }) }],
    });
    windowAggregates.delete(`${previousMinute}`);
  }
}, 60000);

-- dbt tests
version: 2
models:
  - name: orders
    columns:
      - name: id
        tests: [unique, not_null]
      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('users')
              field: id
      - name: total
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0


-- Great Expectations (Python)
import great_expectations as gx

context = gx.get_context()
suite = context.get_expectation_suite('orders_suite')

batch.expect_column_values_to_be_unique('id')
batch.expect_column_values_to_be_between('total', min_value=0)
batch.expect_column_values_to_match_regex('email', r'^[\w.-]+@')

[소규모]
- PostgreSQL → Metabase (직접 쿼리)
- 충분 (< 100K 이벤트/일)

[중규모]
- PostgreSQL + Read Replica
- pg_cron으로 집계 테이블
- Metabase / Superset

[대규모]
- PostgreSQL → Airbyte → BigQuery
- dbt로 변환
- Looker / Tableau
- Kafka로 실시간

[엔터프라이즈]
- Snowflake / Databricks
- Fivetran / Stitch
- dbt Cloud
- Hex / Mode (BI)