Data Engineering & Cloud Platforms

What You Should Understand

• 能畫出一個 modern data platform 的分層架構：sources → ingestion → storage → processing → serving
• 能比較 data lake、data warehouse、lakehouse，並解釋 medallion architecture 和 table formats（Delta / Iceberg / Hudi）
• 知道為什麼 Parquet 是分析工作的預設格式，以及 partitioning 和 predicate pushdown 怎麼省錢
• 能解釋 ETL vs ELT、Airflow DAG 的 idempotency 和 backfill、dbt 的角色
• 理解 Spark 的 driver/executor 模型、shuffle 的成本，並能判斷什麼時候 pandas 就夠了
• 能把 storage / warehouse / processing / streaming / orchestration 對應到 AWS、GCP、Azure 的服務

The Modern Data Platform Landscape

A modern data platform moves data through five logical layers:

直覺：OLTP database 是為了「快速處理單筆交易」設計的（row-oriented、高併發、低延遲），分析查詢卻是「掃很多 rows、只看幾個 columns」— 直接在 production DB 上跑分析會拖垮線上服務，所以要先把資料複製到專為分析設計的系統。這就是整個 data platform 存在的理由。

Who Does What

Where a DS touches each layer:

Data Lake vs Data Warehouse vs Lakehouse

Data warehouse: a system that stores structured, schema-enforced data optimized for SQL analytics (schema-on-write). Data lake: a repository of raw files in open formats on cheap object storage, interpreted at read time (schema-on-read). Lakehouse: data lake storage plus a transactional table format layer, giving warehouse-like guarantees on lake economics.

直覺比喻：warehouse 像整理好的圖書館（上架前先分類編目，找書很快）；data lake 像倉庫（什麼都先丟進去，便宜但要找東西時得自己翻）；lakehouse 是在倉庫裡加上圖書館的目錄系統 — 保留便宜的儲存，補上交易保證和治理。

Medallion Architecture

Lakehouse pipelines are commonly organized into three quality tiers:

直覺：Bronze 保留「事實的原始證據」讓你永遠可以重算；Silver 是「可信的明細」；Gold 是「可以直接給老闆看的數字」。錯誤修復的原則是往上游修 — 如果 Gold 的數字錯了，修 Silver 的邏輯再重跑，而不是手動改 Gold。

Table Formats: Delta Lake / Iceberg / Hudi

Plain Parquet files on S3 have no transaction log — concurrent writes can corrupt each other, and there is no way to update or delete rows safely. Table formats add a metadata layer that provides:

Object Storage Foundations

Why S3-style object storage underpins nearly every data lake:

和傳統 HDFS 或 warehouse 綁定的儲存相比，decoupled compute/storage 是雲端資料架構最核心的轉變：compute 可以隨用隨開隨關，storage 永遠在線且便宜。

Partitioning

Data lake tables are laid out as directory prefixes, e.g. events/dt=2026-07-01/country=TW/part-0001.parquet. Query engines use these prefixes to prune irrelevant files:

• 查詢帶 WHERE dt = '2026-07-01' 時，engine 只列出並讀取那個 prefix 下的檔案 — 掃描量可能從 TB 降到 GB。
• 選 partition key 的原則：低 cardinality、幾乎每個查詢都會 filter 的欄位（date 是最常見的）。用 user_id 這種高 cardinality 欄位做 partition 會產生百萬個小檔案（small-file problem），metadata 操作比讀資料還慢。

File Formats: Parquet vs CSV vs Avro

Predicate pushdown: Parquet 在每個 row group 存了各欄位的 min/max 統計，engine 看到 WHERE amount $>$ 1000 時，會直接跳過 max 小於 1000 的 row groups — 連讀都不讀。加上 column pruning（只讀 SELECT 到的欄位），這就是為什麼同一份資料存 Parquet 的查詢成本常常只有 CSV 的十分之一以下。

直覺：分析查詢通常是「幾百個欄位裡只要 5 個，但要掃很多天」— columnar 格式讓你只付你要的那 5 欄的 I/O。反過來，streaming ingestion 是「一次寫一整筆 record」— row-oriented 的 Avro 比較自然，之後再 compact 成 Parquet。

ETL vs ELT & Orchestration

為什麼 ELT 贏了：cloud warehouse 讓「先全部載入、再用 SQL 慢慢整理」變得便宜，而保留 raw data 意味著 transform 邏輯錯了只要重跑 SQL，不用重新抽取。這催生了 dbt pattern — transformation 全部寫成 SQL SELECT models，用 Git 管版本、寫 tests、自動生成 lineage 文件，把 software engineering 的紀律帶進 SQL 世界。

Orchestration with Airflow-style DAGs

An orchestrator schedules tasks as a DAG (directed acyclic graph) with dependencies, retries, and alerting. Key concepts interviewers probe:

Data Quality Checks in Pipelines

和 ML Pipelines 那頁的 data validation 直接相連 — 在 orchestration 層把品質檢查做成 DAG 中的 blocking tasks：

原則同樣是 fail loud：寧可 dashboard 晚一天，也不要 model 默默用壞資料訓練。

Distributed Processing with Spark

Driver / Executor Model

A Spark application has one driver（解析你的程式、建立執行計畫、調度工作）and many executors（實際持有 data partitions 並執行運算的 worker processes）。資料被切成 partitions 分散在 executors 上平行處理。

Lazy Evaluation and the DAG

Transformations (filter, select, groupBy) do not execute immediately — Spark 只是把它們記錄成一個 logical plan。直到遇上 action（count, write, collect）才會觸發：optimizer 把整條 transformation 鏈整理成 execution DAG（合併運算、把 filter 往資料源推），一次執行。這讓 Spark 能做全局最佳化，但也是新手困惑的來源 — 程式跑到 write 那行才爆錯，錯誤其實在前面的 transformation。

Narrow vs Wide Transformations

Why shuffle is expensive: 所有 executors 把資料按 key 重新分組，寫到 local disk，再透過網路互相交換（all-to-all）。Disk I/O + network + serialization 三重成本，而且是 stage 之間的同步屏障 — 最慢的 partition 拖住所有人。

Partitioning and Skew Tips

When Is pandas Enough?

Batch vs Streaming in the Data Platform

Streaming Ingestion into the Lake

典型 event pipeline：app SDK 發 events → Kafka / Kinesis（durable, ordered, replayable log — producer 和 consumer 解耦）→ 一個 sink job（Kafka Connect / Firehose / Spark Streaming）持續把 events 落到 lake 的 bronze 層 → 下游 batch jobs 接手清洗聚合。訊息系統的 replay 能力很關鍵：consumer 有 bug 時可以回捲 offset 重新消費。

Micro-batch vs True Streaming

大部分「real-time」需求（dashboard 每分鐘更新、feature 5 分鐘內 fresh）micro-batch 就綽綽有餘；毫秒級的 true streaming 只有 fraud、bidding 這類場景才需要。

Freshness vs cost tradeoff：latency 每降一個數量級，架構複雜度和成本大約升一個數量級。面試的加分回答是反問：「這個 metric 晚一小時會造成什麼商業損失？」— 大多數分析用途 daily batch 就夠，streaming 是為明確的 use case 服務，不是預設。

Cloud Big Data Services Map

The same logical layer maps to equivalent managed services on each cloud:

記法：分層記、不要逐一背 — 面試被問「你們是 GCP，我們是 AWS，你 OK 嗎？」時，正確的回答是概念一一對應：BigQuery 對 Redshift、Pub/Sub 對 Kinesis、Composer 對 MWAA，換雲只是換服務名，架構思維完全共通。

Which Database Should I Use? (Workload-driven)

Data Modeling for Analytics

Star Schema

Warehouse 的經典模型：中央的 fact table（事件/交易，數字 measures + 外鍵）連到多個 dimension tables（描述性屬性）。

好處：BI 查詢變成固定模式 — fact join 幾個 dims、GROUP BY 維度、聚合 measures；維度屬性只存一份，改一次全部報表都對。

Grain

Grain = fact table 一列代表什麼（one row per order? per order line? per user per day?）。建模的第一個問題永遠是宣告 grain — grain 不清楚，所有的 join 和聚合都可能默默重複計算。這也是 SQL 面試 fan-out bug 的根源：fact join 到比它 grain 細的表，金額被複製多次，SUM 直接爆掉。

Slowly Changing Dimensions (SCD)

Dimension 屬性會變（user 搬家、改方案），怎麼處理歷史？

SCD Type 2 是面試熱點：查詢時 fact 要 join 在 event_date BETWEEN valid_from AND valid_to 的那一列 dimension，才能還原「事件發生當下」的屬性。用 current 值回算歷史 = 常見的分析錯誤（例如把用戶現在的 premium 身分套用到他還是 free user 時的行為）。

Wide Tables vs Normalized

現代常見折衷：Silver 層維持 normalized star schema 作為 single source of truth，Gold 層用 dbt 物化成 wide tables 給 dashboard 和 ML 直接用 — 兩者不衝突，是不同層的產物。

Real-World Use Cases

Case 1: 建立實驗數據平台（A/B Testing）

你加入一家還在用 Google Sheets 算實驗結果的公司，要建立可信的實驗平台。設計：app 端 SDK 發 exposure 和 conversion events → Kinesis / Kafka → 落地到 S3 bronze（原始 events，append-only, Parquet, partitioned by date）→ 每小時 batch job 清洗成 silver（dedup、joining assignment logs、修正 late events）→ dbt 在 warehouse 建 gold 層 metrics tables（每個 experiment × variant × metric 一列，含 sample size 和 variance）→ 實驗 dashboard 和統計檢定直接讀 gold 表。

關鍵設計決策：exposure logging 要在 randomization 當下記錄（避免 selection bias）；metrics 定義集中在 dbt models（避免每個 DS 自己算一套 CTR）；bronze 永遠保留 raw events，實驗定義改了可以 backfill 重算。

Interview follow-ups:

• 為什麼 events 要先落 lake 再進 warehouse，而不是直接寫進 warehouse？（成本、replay 能力、非結構化欄位的彈性、warehouse 不適合高頻小筆寫入）
• Late-arriving events（手機離線補傳）怎麼處理？（watermark + 每天重跑最近 N 天的 partitions）
• 怎麼確保兩個 DS 算同一個實驗得到相同數字？（single source of truth 的 metrics 層 — 這就是 gold 層存在的理由）

Case 2: Feature Pipeline 從 pandas 遷移到 Spark 的時機

你的 churn model feature pipeline 是一個每天跑的 pandas script，讀 warehouse 匯出的 CSV。一年後資料從 2GB 長到 60GB，script 開始 OOM、跑 6 小時。你要決定：加大機器記憶體？改寫成 Spark？還是把邏輯下推到 warehouse SQL / dbt？

分析框架：（1）60GB 其實還在「大機器 + Polars/DuckDB」能處理的範圍，最便宜的修法可能是換格式（CSV → Parquet）+ 換 engine，成本一天就能完成；（2）如果成長趨勢明確會到 TB 級、或需要和 lake 上的 event 資料 join，才值得付出 Spark 的複雜度；（3）如果邏輯全是 SQL 可表達的聚合，下推到 warehouse 用 dbt 管理可能是最好解 — 免維運、有 lineage、和其他表共用品質檢查。遷移到 Spark 時的重點：用 broadcast join 處理小 dimension、確認 shuffle partitions 設定、每天 overwrite 對應的 feature partition 保持 idempotent。

Interview follow-ups:

• 你怎麼驗證遷移後 features 和舊 pipeline 一致？（同一天資料雙跑、逐欄位比對統計量 — 呼應 train-serving skew）
• 什麼訊號告訴你「該上 Spark 了」？（資料超過單機記憶體數倍、需要跨 TB join、單機跑不進 SLA）
• 為什麼 CSV 換 Parquet 就能救回一大段效能？（columnar + 壓縮 + 只讀需要的欄位）

Case 3: 選 BigQuery 還是 Self-host？

新創公司（8 人 data 團隊）要選分析平台：BigQuery（serverless, pay-per-query）還是自建（ClickHouse on Kubernetes）。決策維度：

給小團隊的合理建議：先用 managed（BigQuery），把工程時間花在產生商業價值的分析上；等 query 帳單成長到「一個工程師年薪」的量級，才值得評估 self-host — 而且因為資料存在開放格式的 lake（Iceberg/Parquet），遷移時搬的是 compute 不是資料，lock-in 有限。

Interview follow-ups:

• Pay-per-query 模式下，怎麼防止一個失控查詢燒掉預算？（partition filter 強制、查詢 byte 上限、按 team 設 quota）
• 什麼情況 self-host 明顯划算？（查詢量極大且穩定、latency 需求特殊、合規要求資料不出機房）
• 怎麼降低對單一雲的 lock-in？（open table formats、SQL 標準化、把 transform 邏輯放 dbt 而不是 proprietary 工具）

Hands-on: Data Engineering in Python

PySpark: Read, Aggregate, Write

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("daily_user_features").getOrCreate()

# Read only one day via partition pruning (dt is a partition column)
events = (
    spark.read.parquet("s3://lake/silver/events/")
    .filter(F.col("dt") == "2026-07-01")          # partition pruning: reads one prefix
    .select("user_id", "event_type", "amount")     # column pruning: Parquet reads 3 columns
)

# Narrow transformation (no shuffle)
purchases = events.filter(F.col("event_type") == "purchase")

# Wide transformation: groupBy triggers a shuffle by user_id
user_features = purchases.groupBy("user_id").agg(
    F.count("*").alias("purchase_cnt_1d"),
    F.sum("amount").alias("purchase_amt_1d"),
    F.avg("amount").alias("avg_order_value_1d"),
)

# Broadcast join: small dimension table copied to every executor (no shuffle)
users = spark.read.parquet("s3://lake/silver/dim_user/")
enriched = user_features.join(F.broadcast(users), "user_id", "left")

# Control output file count, then overwrite ONE partition (idempotent)
(
    enriched.repartition(8)
    .write.mode("overwrite")
    .parquet("s3://lake/gold/user_features/dt=2026-07-01/")
)

Airflow DAG Sketch

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta

default_args = {
    "retries": 2,                              # transient failures auto-recover
    "retry_delay": timedelta(minutes=10),
}

with DAG(
    dag_id="daily_user_features",
    start_date=datetime(2026, 1, 1),
    schedule="0 3 * * *",                      # daily at 03:00, one run per logical date
    catchup=True,                              # enables backfill of past dates
    default_args=default_args,
) as dag:

    # Sensor: wait until upstream events table for this date is ready
    wait_for_events = ExternalTaskSensor(
        task_id="wait_for_events",
        external_dag_id="ingest_events",
        timeout=3600,
    )

    # Each task processes exactly one date partition -> idempotent reruns
    build_features = PythonOperator(
        task_id="build_features",
        python_callable=run_spark_job,          # submits the PySpark job above
        op_kwargs={"ds": "{{ ds }}"},           # logical date injected by Airflow
    )

    validate = PythonOperator(
        task_id="validate_output",
        python_callable=check_row_count_and_nulls,  # fail loud -> blocks downstream
        op_kwargs={"ds": "{{ ds }}"},
    )

    wait_for_events >> build_features >> validate

dbt-style SQL Model

-- models/gold/fct_daily_user_metrics.sql
-- dbt manages dependencies via ref(); materialized as an incremental table

{{ config(materialized="incremental", unique_key="user_date_key") }}

with purchases as (

    select
        user_id,
        date(event_ts)  as event_date,
        amount
    from {{ ref("stg_events") }}               -- silver-layer staging model
    where event_type = 'purchase'
    {% if is_incremental() %}
      and date(event_ts) >= dateadd(day, -3, current_date)  -- reprocess late data
    {% endif %}

)

select
    user_id || '_' || event_date as user_date_key,   -- declared grain: user x day
    user_id,
    event_date,
    count(*)     as purchase_cnt,
    sum(amount)  as purchase_amt
from purchases
group by user_id, event_date

Interview Signals

What interviewers listen for:

• 你能畫出資料從 OLTP source 到 dashboard/model 的完整路徑，並說出每一層的職責
• 你會先問資料量、latency 需求和成長速度，再選工具 — 而不是所有東西都上 Spark 或 streaming
• 你主動提到 idempotency、backfill、partition overwrite 這些 pipeline 可靠性的關鍵字
• 你能解釋 Parquet + partitioning 為什麼省錢（column pruning、predicate pushdown、partition pruning）
• 你把雲服務當成分層概念的實作（BigQuery ≈ Redshift ≈ Synapse），換雲不換架構思維

Practice

Flashcards

Quiz