• 수집층: Kafka Connect / HTTP Ingest / DB CDC
• 저장층: 오브젝트 스토리지(Parquet/ORC), 파일 스토리지(NFS)
• 처리층: Spark(배치), Flink(스트림), SQL 엔진
• 카탈로그: 메타스토어/스키마 레지스트리
• 거버넌스: IAM/정책/암호화/감사
• 오케스트레이션: Airflow/워크플로

[Producers]
  └─(CDC/Kafka/HTTP)──▶  [Ingest]
                           └─▶  [Raw Zone (S3 호환)]
                                   └─▶  [ETL/ELT: Spark/Flink]
                                          └─▶  [Curated Zone]
                                                  └─▶  [Catalog/BI/AI]

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

with DAG("daily_lake_etl", start_date=datetime(2025,1,1), schedule_interval="@daily", catchup=False) as dag:
  extract = BashOperator(task_id="extract", bash_command="python3 extract_rdbms.py {{ ds }}")
  transform = SparkSubmitOperator(task_id="transform", application="etl/jobs/curate_parquet.py", application_args=["--date", "{{ ds }}"]) 
  load = BashOperator(task_id="load", bash_command="python3 load_dw.py {{ ds }}")
  extract >> transform >> load

-- 파티션드 테이블 예시
CREATE TABLE lake.sales_curated (
  date DATE, region STRING, sku STRING, qty INT, amount DOUBLE
) USING PARQUET
PARTITIONED BY (date)
LOCATION 's3://lake/curated/sales';

INSERT OVERWRITE TABLE lake.sales_curated
SELECT date, region, sku, SUM(qty), SUM(amount)
FROM lake.raw_sales
WHERE date = '${date}'
GROUP BY date, region, sku;

CREATE TABLE clicks (
  user_id STRING, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (...);

CREATE TABLE agg AS
SELECT window_start, window_end, user_id, COUNT(*) AS c
FROM TABLE(
  TUMBLE(TABLE clicks, DESCRIPTOR(ts), INTERVAL '1' MINUTE)
)
GROUP BY window_start, window_end, user_id;

{
  "connector.class": "io.confluent.connect.s3.S3SinkConnector",
  "topics": "events",
  "s3.bucket.name": "lake/raw/events",
  "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
  "flush.size": 100000
}