大數據處理與上億筆資料

問題拆解：面試官到底在問什麼 高頻

「把兩份資料 merge 在一起」看似一題簡單的 coding 題，其實是一個可以無限往下挖的系統題。面試官通常會先讓你寫出能動的版本（兩個 List[dict] 依 id 合併），然後一層層加條件：資料變大、記憶體不夠、要每天跑、要即時、資料有衝突、key 分布不均……每一層都在測試不同能力：

開始寫之前要先問的問題

1. Join key 是什麼？是 user_id 還是 email / device id？兩邊的 key 格式是否一致（大小寫、前後空白、hash 過的 email）？
2. Key 在單一來源中是否唯一？同一個 id 在一份資料內出現多次時，要取最新還是合併？
3. Join 類型？inner（只要兩邊都有的）、left（以主資料為準）、full outer（全部保留）？使用者屬性整合通常是 full outer。
4. 欄位衝突規則？同一欄位兩邊值不同時：最新時間戳優先、來源優先序、非空值優先、還是 list 取聯集？
5. 規模與資源？筆數、每筆大小、機器記憶體、能接受的執行時間、是一次性還是每天跑。
6. 輸出要給誰用？寫回 DB 給線上服務查（需要低延遲 key-value 查詢）還是寫成檔案給分析（需要列式格式）？

Level 0–1：記憶體放得下 —— Hash Map 合併 高頻 基礎

Appier 真實考題之一：「合併兩個 List[dict] 中相同 id 的資料」以及「合併多個資料來源的 user attributes」。最直接的解法是用 dict 當 hash map：以 id 為 key 掃過所有來源，時間複雜度 O(n + m)（n、m 為兩邊筆數，假設每筆欄位數為常數），空間 O(n + m)。

為什麼不能用雙層迴圈

雙層迴圈對每一筆 A 去掃整個 B 找相同 id，是 O(n × m)。兩邊各 10 萬筆就是 10¹⁰ 次比較，在 Python 裡要跑好幾個小時；hash map 版本只需約 20 萬次 dict 操作，不到一秒。這就是資料庫中 nested loop join vs hash join 的差別。

基本版：依 id 合併，後者覆蓋前者

from typing import Any

def merge_by_id(a: list[dict[str, Any]], b: list[dict[str, Any]],
                key: str = "id") -> list[dict[str, Any]]:
    """Full outer merge: 兩邊都保留，同 id 的欄位合併，b 的值覆蓋 a。"""
    merged: dict[Any, dict[str, Any]] = {}
    for rec in a:
        merged.setdefault(rec[key], {}).update(rec)
    for rec in b:
        merged.setdefault(rec[key], {}).update(rec)
    return list(merged.values())   # Python 3.7+ dict 保留插入順序

a = [{"id": 1, "name": "Amy"}, {"id": 2, "name": "Bob"}]
b = [{"id": 2, "email": "bob@x.com"}, {"id": 3, "name": "Cat"}]
print(merge_by_id(a, b))
# [{'id': 1, 'name': 'Amy'}, {'id': 2, 'name': 'Bob', 'email': 'bob@x.com'}, {'id': 3, 'name': 'Cat'}]

進階版：可設定的欄位衝突規則

真實的 user attributes 整合（例如 CRM、App SDK、網站事件、線下 POS）一定會有衝突。好的答案是把規則參數化，常見策略：

Latest wins（最新時間戳優先）

每個欄位帶 updated_at，取最新的值。適合會變動的屬性（地址、會員等級）。注意要比較的是欄位層級還是整筆記錄層級的時間戳。

Source priority（來源優先序）

例如 CRM（使用者自己填的）> App SDK > 第三方推論。適合可信度不同的來源。

Non-null wins（非空值優先）

只有在新值非空時才覆蓋，避免空值洗掉資料。通常與上面兩種規則組合。

Union / Aggregate（聚合）

興趣標籤、裝置列表取聯集；購買金額相加；首次造訪時間取最小值。

from collections.abc import Iterable, Callable
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

SOURCE_PRIORITY = {"crm": 3, "app_sdk": 2, "web": 1, "third_party": 0}

@dataclass
class FieldValue:
    value: Any
    ts: datetime
    source: str

def is_empty(v: Any) -> bool:
    return v is None or v == "" or v == [] or v == {}

# 每種規則都是 (舊值, 新值) -> 勝出值
def latest_wins(old: FieldValue, new: FieldValue) -> FieldValue:
    if is_empty(new.value):
        return old
    if is_empty(old.value):
        return new
    # 同時間戳時用來源優先序打破平手，確保結果與輸入順序無關（deterministic）
    return max(old, new, key=lambda f: (f.ts, SOURCE_PRIORITY.get(f.source, -1)))

def priority_wins(old: FieldValue, new: FieldValue) -> FieldValue:
    if is_empty(new.value):
        return old
    if is_empty(old.value):
        return new
    return max(old, new, key=lambda f: (SOURCE_PRIORITY.get(f.source, -1), f.ts))

def union_list(old: FieldValue, new: FieldValue) -> FieldValue:
    merged = sorted(set(old.value or []) | set(new.value or []))
    return FieldValue(merged, max(old.ts, new.ts), "merged")

def min_value(old: FieldValue, new: FieldValue) -> FieldValue:
    return min(old, new, key=lambda f: f.value)

RULES: dict[str, Callable[[FieldValue, FieldValue], FieldValue]] = {
    "email": priority_wins,        # 使用者自填的最可信
    "city": latest_wins,           # 會搬家，取最新
    "interests": union_list,       # 標籤取聯集
    "first_seen": min_value,       # 首次出現時間取最小
}
DEFAULT_RULE = latest_wins

def merge_user_attributes(sources: dict[str, Iterable[dict]],
                          key: str = "user_id") -> dict[Any, dict[str, Any]]:
    """sources: {"crm": [...], "app_sdk": [...]}；每筆記錄含 key 與 updated_at。"""
    state: dict[Any, dict[str, FieldValue]] = {}
    for source, records in sources.items():
        for rec in records:
            uid = str(rec[key]).strip().lower()           # key 正規化
            ts = rec.get("updated_at") or datetime.min
            user = state.setdefault(uid, {})
            for f, v in rec.items():
                if f in (key, "updated_at"):
                    continue
                new = FieldValue(v, ts, source)
                old = user.get(f)
                user[f] = new if old is None else RULES.get(f, DEFAULT_RULE)(old, new)
    # 攤平成一般 dict 輸出；可另外保留 lineage（值來自哪個來源）方便除錯
    return {uid: {f: fv.value for f, fv in fields.items()} for uid, fields in state.items()}

記憶體估算：1 億筆到底要多少記憶體

面試官問「上億筆怎麼辦」時，第一件事是算一下，而不是直接跳到 Spark。估算分兩部分：原始資料大小與程式中的物件大小。

import sys
rec = {"user_id": "u_000123456", "city": "Taipei", "age": 31, "gender": "F", "tier": "gold"}
print(sys.getsizeof(rec))                                  # dict 本身（不含 key/value 物件）
deep = sys.getsizeof(rec) + sum(sys.getsizeof(k) + sys.getsizeof(v) for k, v in rec.items())
print(deep)   # 實測會是數百 bytes；key 字串在多筆間通常是共用的 interned 物件，但 value 不是

# 更省記憶體的表示法
from dataclasses import dataclass
@dataclass(slots=True)          # Python 3.10+；slots 讓每個 instance 沒有 __dict__
class User:
    user_id: str
    city: str
    age: int
# 或者用 tuple / numpy structured array / pandas（底層為連續陣列）/ Apache Arrow

L1 技巧：小表建 hash、大表串流

如果 A 有 100 萬筆（放得下）、B 有 1 億筆（放不下），只需把 A 建成 dict，然後用 generator 一行一行讀 B 並輸出結果。記憶體只與 A 成正比。這就是資料庫 hash join 的 build side / probe side，也是 Spark broadcast join 的單機版本。

import csv
from collections.abc import Iterator

def stream_csv(path: str) -> Iterator[dict]:
    with open(path, newline="", encoding="utf-8") as f:
        yield from csv.DictReader(f)          # 一次只在記憶體中保留一列

def header(path: str) -> list[str]:
    with open(path, newline="", encoding="utf-8") as f:
        return next(csv.reader(f))

def hash_join_stream(small_path: str, big_path: str, out_path: str, key: str = "user_id") -> None:
    """Left join：以大表為主，小表的屬性補進來；大表的非空值優先。"""
    build = {row[key]: row for row in stream_csv(small_path)}       # build side：小表
    fields = list(dict.fromkeys(header(big_path) + header(small_path)))   # 保序去重
    with open(out_path, "w", newline="", encoding="utf-8") as out:
        writer = csv.DictWriter(out, fieldnames=fields)
        writer.writeheader()
        for row in stream_csv(big_path):                              # probe side：大表
            match = build.get(row[key], {})
            writer.writerow({**match, **{k: v for k, v in row.items() if v != ""}})

Level 2：單機記憶體放不下 —— Out-of-core 演算法 高頻 進階

當兩邊都放不進記憶體（例如 8GB RAM、兩份各 30GB 資料），核心想法只有一個：把資料切成記憶體放得下的小塊，並讓需要配對的資料落在同一塊。有兩種切法：依 key 排序（sort-based）或依 key 的 hash 分桶（hash-based）。這兩種方法分別對應資料庫的 sort-merge join 與 grace hash join，也正是 MapReduce / Spark shuffle 的底層原理。

情況一：兩個檔案都已依 key 排序 —— 串流 merge

如果兩份檔案都已依 user_id 排序（例如上游匯出時已排序，或是 key 遞增的 log），可以像 merge sort 的 merge 步驟一樣兩個指標同時往前走，記憶體只需 O(同一 key 的筆數)，時間 O(n + m)，而且只需循序讀（sequential I/O，對磁碟最友善）。

import json
from collections.abc import Iterator
from itertools import groupby

def read_jsonl(path: str) -> Iterator[dict]:
    with open(path, encoding="utf-8") as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

def grouped(path: str, key: str) -> Iterator[tuple[str, list[dict]]]:
    """把連續相同 key 的記錄聚成一組（前提：檔案已依 key 排序）。"""
    for k, recs in groupby(read_jsonl(path), key=lambda r: r[key]):
        yield k, list(recs)

def sorted_merge_join(path_a: str, path_b: str, out_path: str, key: str = "user_id") -> None:
    """Full outer sort-merge join。"""
    ga, gb = grouped(path_a, key), grouped(path_b, key)
    a, b = next(ga, None), next(gb, None)
    with open(out_path, "w", encoding="utf-8") as out:
        def emit(recs: list[dict]) -> None:
            merged: dict = {}
            for r in recs:                      # 這裡可換成上一節的衝突規則
                merged.update({k: v for k, v in r.items() if v is not None})
            out.write(json.dumps(merged, ensure_ascii=False) + "\n")

        while a is not None or b is not None:
            if b is None or (a is not None and a[0] < b[0]):
                emit(a[1]); a = next(ga, None)            # 只有 A 有
            elif a is None or b[0] < a[0]:
                emit(b[1]); b = next(gb, None)            # 只有 B 有
            else:
                emit(a[1] + b[1])                         # 兩邊都有：合併
                a, b = next(ga, None), next(gb, None)

情況二：檔案未排序 —— External Merge Sort

外部排序分兩階段：

1. Run generation（切塊排序）：每次讀入記憶體放得下的量（例如 1GB），在記憶體內排序後寫成一個暫存檔（稱為 run）。30GB 資料 → 約 30 個已排序的 run。
2. K-way merge（多路合併）：同時開啟所有 run，用一個大小為 k 的 min-heap 每次取出最小的元素寫出。記憶體只需 O(k) 個元素加上每個檔案的讀取緩衝區。Python 的 heapq.merge 正是 lazy 的 k-way merge。

複雜度：比較次數 O(N log N)；I/O 方面，若 run 數量 k 不超過可同時開啟的檔案數與緩衝區限制，只需讀寫資料兩趟（一趟產生 run、一趟合併）。若 run 太多，則需要多輪合併（multi-pass），趟數約為 log_k(run 數)。

import heapq
import json
import os
import tempfile
from collections.abc import Iterator
from itertools import islice

def _write_run(records: list[dict], key: str, tmpdir: str) -> str:
    records.sort(key=lambda r: r[key])
    fd, path = tempfile.mkstemp(dir=tmpdir, suffix=".run")
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    return path

def _iter_run(path: str) -> Iterator[dict]:
    with open(path, encoding="utf-8", buffering=1 << 20) as f:     # 1MB 讀取緩衝
        for line in f:
            yield json.loads(line)

def external_sort(in_path: str, out_path: str, key: str,
                  chunk_records: int = 1_000_000) -> None:
    """chunk_records 依記憶體調整：每塊在記憶體中的 Python 物件大小要遠小於可用 RAM。"""
    with tempfile.TemporaryDirectory() as tmpdir:
        runs: list[str] = []
        with open(in_path, encoding="utf-8") as f:
            while True:
                lines = list(islice(f, chunk_records))
                if not lines:
                    break
                runs.append(_write_run([json.loads(l) for l in lines], key, tmpdir))
        # heapq.merge 是 lazy 的：任何時刻每個 run 只有一筆在 heap 中
        merged = heapq.merge(*(_iter_run(p) for p in runs), key=lambda r: r[key])
        with open(out_path, "w", encoding="utf-8") as out:
            for r in merged:
                out.write(json.dumps(r, ensure_ascii=False) + "\n")

# 兩份都外部排序後，再呼叫上面的 sorted_merge_join 即可完成上億筆的 join：
# external_sort("a.jsonl", "a.sorted", "user_id")
# external_sort("b.jsonl", "b.sorted", "user_id")
# sorted_merge_join("a.sorted", "b.sorted", "merged.jsonl")

情況三：Hash Partition 分桶 —— Grace Hash Join

如果不需要排序結果，只要 join，hash 分桶通常比排序更快（不需要 O(N log N) 比較）：

1. Partition 階段：讀 A，對每筆算 bucket = hash(key) % N，寫到 a_bucket_{i}；B 用同一個 hash 函式與同一個 N 寫到 b_bucket_{i}。相同 key 必定落在相同編號的桶。
2. Join 階段：對每個 i，把 a_bucket_i 載入記憶體建 dict，串流讀 b_bucket_i 做 hash join。每桶獨立，天然可平行（多 process 或多台機器）。

N 怎麼選？讓每一桶的較小一邊能放進記憶體。例如 A 載入記憶體後估計 60GB，可用 4GB 給 hash table，N 至少 15，保守取 64 或 128（留餘裕給 skew）。但 N 也受限於同時開啟的檔案數（ulimit -n，常見預設 1024）與每個檔案的寫入緩衝。

import json
import zlib
from contextlib import ExitStack
from collections.abc import Iterator

def stable_bucket(key: str, n: int) -> int:
    # 不要用內建 hash()：str 的 hash 每個 process 隨機化（PYTHONHASHSEED），
    # 跨 process / 跨次執行結果不同，A 與 B 會分到不同桶！
    return zlib.crc32(key.encode("utf-8")) % n

def partition(in_path: str, prefix: str, key: str, n: int) -> None:
    with ExitStack() as stack, open(in_path, encoding="utf-8") as f:
        outs = [stack.enter_context(open(f"{prefix}_{i:04d}.jsonl", "w", encoding="utf-8"))
                for i in range(n)]
        for line in f:
            rec = json.loads(line)
            outs[stable_bucket(str(rec[key]), n)].write(line)

def join_bucket(i: int, key: str) -> Iterator[dict]:
    """Full outer merge 單一桶：每桶已經夠小，兩邊都載入同一個 dict。
    同一 key 在 A 或 B 內出現多次也會被合併成一筆；B 的非空值覆蓋 A。
    （若 B 的 key 保證唯一，也可只載入 A 當 build side、串流讀 B 當 probe side。）"""
    merged: dict[str, dict] = {}
    for side in ("a", "b"):
        with open(f"{side}_{i:04d}.jsonl", encoding="utf-8") as f:
            for line in f:
                r = json.loads(line)
                cur = merged.setdefault(str(r[key]), {})
                cur.update({f: v for f, v in r.items() if v is not None})
    yield from merged.values()

def grace_hash_join(a_path: str, b_path: str, out_path: str,
                    key: str = "user_id", n: int = 64) -> None:
    partition(a_path, "a", key, n)
    partition(b_path, "b", key, n)
    with open(out_path, "w", encoding="utf-8") as out:
        for i in range(n):                  # 可改用 multiprocessing.Pool 平行處理各桶
            for rec in join_bucket(i, key):
                out.write(json.dumps(rec, ensure_ascii=False) + "\n")

Join 策略總整理：Hash / Sort-merge / Broadcast / Shuffle 高頻 進階

單機的演算法搬到分散式環境後，多了一個成本最高的動作：網路傳輸（shuffle）。分散式 join 的核心問題變成「怎麼讓相同 key 的資料到同一台機器上，並且搬最少的資料」。

Nested loop join

對外表每一筆掃內表，O(n×m)。只在其中一邊很小、或是非等值條件（例如 a.ts BETWEEN b.start AND b.end）時使用。Spark 中對應 BroadcastNestedLoopJoin / CartesianProduct，看到它通常代表 join 條件寫錯了。

Hash join

小的一邊建 hash table（build），大的一邊探查（probe）。O(n+m)，只支援等值 join。

Sort-merge join

兩邊依 key 排序後合併。可處理很大的資料（排序可以 spill 到磁碟），是 Spark 對兩張大表做等值 join 時的預設策略。

Broadcast join（map-side join）

把小表完整複製到每個 executor，大表完全不用 shuffle，每個 partition 在本地做 hash join。Spark 中小於 spark.sql.autoBroadcastJoinThreshold（預設 10MB）的表會自動 broadcast，也可以用 broadcast() hint 強制。

Shuffle join

兩邊都依 hash(key) % P 重新分區，透過網路送到對應的節點，之後在每個分區做 sort-merge 或 hash join。成本：兩張表都要經過網路和磁碟。

Bucketed join（storage-partitioned join）

如果兩張表寫入時就已用相同 key、相同桶數做 bucketing，join 時可以跳過 shuffle。適合天天重複 join 的大表，以寫入時的成本換讀取時的效能。

資料傾斜（skew）與 salting

Shuffle join 的效能取決於最慢的那個 task。若某個 key（例如熱門 app id、某個大客戶、null key）佔了 30% 的資料，那個 key 所在的 task 會跑非常久（Spark UI 上看到 200 個 task 中 199 個幾秒完成、1 個跑 40 分鐘），甚至 OOM。

Salting 的做法：把熱點 key 在大表那一側加上隨機後綴 key_0 … key_{S-1}，打散到 S 個 task；小表（或另一側）那一邊則把該 key複製 S 份，每份配上一個後綴，確保仍能配對到。代價是另一側資料量變成 S 倍，所以通常只對熱點 key 做 salting。

from pyspark.sql import functions as F

SALT = 16
hot_keys = ["app_123", "app_456"]              # 可先 groupBy(key).count() 找出熱點

events = events.withColumn(
    "salt",
    F.when(F.col("app_id").isin(hot_keys), (F.rand(seed=42) * SALT).cast("int"))
     .otherwise(F.lit(0)))
events = events.withColumn("join_key", F.concat_ws("#", "app_id", "salt"))

# 維度表：熱點 key 複製 SALT 份，其他 key 只要 salt=0
salts = spark.range(SALT).withColumnRenamed("id", "salt")
dim_hot = dim.filter(F.col("app_id").isin(hot_keys)).crossJoin(salts)
dim_cold = dim.filter(~F.col("app_id").isin(hot_keys)).withColumn("salt", F.lit(0).cast("long"))
dim_salted = dim_hot.unionByName(dim_cold) \
    .withColumn("join_key", F.concat_ws("#", "app_id", "salt"))

joined = events.join(dim_salted.drop("app_id", "salt"), "join_key")

MapReduce 模型與 Hadoop 生態系 進階

MapReduce（Google 2004 論文，Hadoop 為其開源實作）把「上一節的 hash 分桶」推廣到上千台機器：使用者只寫 map 與 reduce 兩個純函式，框架負責切分資料、排程、shuffle、容錯。雖然現在新專案多半用 Spark，但 Spark 的 shuffle、stage、partition 概念都源自 MapReduce，面試很常從這裡問起。

執行流程

1. Input split：輸入檔案依 HDFS block（預設 128MB）切成 split，每個 split 啟動一個 map task，盡量排在存有該 block 的機器上（data locality：搬運算而不是搬資料）。
2. Map：每筆記錄 → 0 到多個 (key, value)。輸出先寫進記憶體環形緩衝區，滿了就 spill 到本地磁碟。
3. Partition + Sort（map 端）：依 hash(key) % R 決定要送給哪個 reducer，並在每個分區內依 key 排序。可選的 combiner 在這裡先做局部聚合。
4. Shuffle：每個 reducer 透過網路從所有 mapper 拉取屬於自己的分區（M × R 次傳輸）。
5. Merge + Reduce：reducer 對拉來的已排序片段做 merge（就是外部排序的 k-way merge），相同 key 的值聚在一起交給 reduce(key, values)。
6. Output：結果寫回 HDFS（會被複製 3 份）。

經典範例：Word Count（用 Python 模擬）

from collections import defaultdict
from collections.abc import Iterator
from itertools import groupby
import re

def mapper(line: str) -> Iterator[tuple[str, int]]:
    for word in re.findall(r"\w+", line.lower()):
        yield word, 1

def combiner(word: str, counts: list[int]) -> tuple[str, int]:
    return word, sum(counts)          # 與 reducer 相同：sum 滿足交換律、結合律

def reducer(word: str, counts: Iterator[int]) -> tuple[str, int]:
    return word, sum(counts)

def run_mapreduce(splits: list[list[str]], num_reducers: int = 2) -> dict[str, int]:
    # 1) map + combine，並依 hash 分到 reducer 分區
    partitions: list[list[tuple[str, int]]] = [[] for _ in range(num_reducers)]
    for split in splits:                         # 每個 split 是一個 map task
        local: dict[str, list[int]] = defaultdict(list)
        for line in split:
            for k, v in mapper(line):
                local[k].append(v)
        for k, vs in local.items():
            ck, cv = combiner(k, vs)
            partitions[hash(ck) % num_reducers].append((ck, cv))   # 單一 process 內 hash 一致即可
    # 2) shuffle 完成後，每個 reducer 對自己的分區排序 + 分組 + reduce
    result: dict[str, int] = {}
    for part in partitions:
        part.sort(key=lambda kv: kv[0])
        for k, group in groupby(part, key=lambda kv: kv[0]):
            rk, rv = reducer(k, (v for _, v in group))
            result[rk] = rv
    return result

print(run_mapreduce([["the cat sat", "the dog"], ["the end"]]))
# {'cat': 1, 'dog': 1, ...  'the': 3}（輸出順序依分區而定）

Combiner 只能用在滿足交換律與結合律的運算（sum、count、max、min）。平均值不能直接 combine：avg(avg(1,2), avg(3)) ≠ avg(1,2,3)；正確做法是 combine 成 (sum, count)，最後再相除。

用 MapReduce 做兩份資料的 join（reduce-side join）

map 階段對兩份資料都輸出 (user_id, (source_tag, record))；shuffle 把同一 user_id 的記錄送到同一個 reducer；reduce 階段依 tag 分成 A、B 兩組再做合併。這就是 Spark shuffle join 的原型。若一邊很小，則可做 map-side join：把小表放進 Distributed Cache 發給每個 mapper，跳過 shuffle（即 broadcast join）。

為什麼 MapReduce 慢

• 每個 job 之間都落地到 HDFS：多步驟的計算（例如迭代式 ML、多次 join）要串多個 job，中間結果每次都寫 HDFS（含 3 份複製）再讀回來。
• Map 輸出 spill 到磁碟、shuffle 強制排序：即使不需要排序也要排。
• JVM 啟動與排程開銷：每個 task 啟動成本高，小 job 也要數十秒。
• 程式模型僵硬：只有 map/reduce，複雜邏輯要拆成很多 job，難以全域最佳化。

Spark 的改進：用 DAG 把多個步驟合併最佳化、中間結果盡量留在記憶體、只在真正需要 shuffle 的地方切 stage、executor 長駐（不必每個 task 重啟 JVM）。

HDFS 架構

讀取流程：client 問 NameNode 拿 block 位置 → 直接向最近的 DataNode 讀取（資料不經過 NameNode）。寫入流程：NameNode 分配 DataNode → client 對 DataNode 組成 pipeline 逐一轉寫 → ack 回傳。HDFS 設計為 write-once-read-many：支援 append，不支援隨機修改，適合大檔案循序讀，不適合低延遲隨機存取。

YARN：叢集資源管理

ResourceManager

全叢集的資源仲裁者，負責排程（FIFO / Capacity / Fair scheduler）。

NodeManager

每台機器一個，管理本機的 container（CPU/記憶體配額），回報資源使用。

ApplicationMaster

每個應用一個（例如一個 Spark job 的 driver 可以跑在 AM 中），向 RM 申請 container、監控自己的 task。

Container

資源配額單位（例如 4 vCore + 8GB）。Spark executor 就跑在 container 裡。

現在 Spark 也常直接跑在 Kubernetes 上（spark-on-k8s），概念對應：driver/executor 是 pod，資源由 K8s scheduler 分配。

Hive 與 HiveQL

Hive 讓分析師用 SQL（HiveQL）查詢 HDFS 上的檔案：SQL 被編譯成 MapReduce / Tez / Spark 的執行計畫。它最重要的遺產是 Hive Metastore：記錄「表名 → schema、檔案位置、分區」的中央目錄，Spark、Presto/Trino 等引擎都能共用。Hive 是 schema-on-read：資料先放進去，讀的時候才套用 schema。

-- 外部表：刪表不會刪檔案；依日期分區、依 user_id 分桶
CREATE EXTERNAL TABLE user_events (
  user_id     STRING,
  event_name  STRING,
  properties  MAP<STRING, STRING>,
  event_ts    TIMESTAMP
)
PARTITIONED BY (dt STRING)
CLUSTERED BY (user_id) INTO 256 BUCKETS
STORED AS PARQUET
LOCATION 's3://my-bucket/warehouse/user_events/';

-- 分區裁剪（partition pruning）：只會掃 dt='2026-11-01' 那個目錄
SELECT event_name, COUNT(DISTINCT user_id) AS uu
FROM user_events
WHERE dt = '2026-11-01'
GROUP BY event_name;

-- 覆寫單一分區：重跑同一天結果相同（idempotent）
INSERT OVERWRITE TABLE daily_user_profile PARTITION (dt = '2026-11-01')
SELECT user_id, MAX(event_ts) AS last_seen, COUNT(*) AS events
FROM user_events
WHERE dt = '2026-11-01'
GROUP BY user_id;

Spark 核心：RDD、DataFrame、DAG 與執行模型 高頻 深入

Spark 是目前最主流的分散式批次運算引擎。面試的深度通常會到：為什麼比 MapReduce 快、lazy evaluation、narrow/wide dependency、stage 怎麼切、shuffle 為什麼貴、怎麼調 partition、cache 何時用。

RDD vs DataFrame vs Dataset

Lazy evaluation、DAG、Job / Stage / Task

Spark 的操作分兩類：transformation（select、filter、join、groupBy……只建立計畫，不執行）與 action（count、collect、write、show……觸發真正執行）。Lazy 的好處：看到整個計畫後才最佳化（例如把 filter 下推到讀檔階段、合併多個 map），而且只計算 action 真正需要的部分。

Job

每個 action 觸發一個（或多個）job。

Stage

Job 依 shuffle 邊界（wide dependency）切成多個 stage；stage 內的操作可以 pipeline 串在一起，不落地。

Task

Stage 中每個 partition 一個 task，是排程到 executor core 上的最小單位。Task 數 = 該 stage 的 partition 數。

Narrow dependency

子 partition 只依賴父的少數固定 partition（map、filter、union、broadcast join）。不需 shuffle，可 pipeline，失敗時只需重算少數 partition。

Wide dependency

子 partition 依賴父的所有 partition（groupByKey、reduceByKey、join（非 co-partitioned）、distinct、repartition）。需要 shuffle：上游寫 shuffle file 到本地磁碟，下游跨網路拉取。

Shuffle 為什麼貴

• 磁碟 I/O：map 端把輸出依目標 partition 寫成 shuffle file（含排序與 spill）。
• 網路 I/O：每個下游 task 要從所有上游 executor 拉資料，M × R 個連線/區塊。
• 序列化/反序列化與壓縮：CPU 成本。
• 記憶體壓力：下游聚合/排序時可能 spill 或 OOM。
• 容錯：若存放 shuffle file 的 executor 掛了，需要重跑上游 stage（fetch failed）。

減少 shuffle 的手段：broadcast join、用 reduceByKey 而非 groupByKey（前者有 map 端局部聚合，類似 combiner）、提早 filter / select 減少欄位與列數、bucketing、避免不必要的 repartition / distinct / orderBy。

# RDD 時代的經典比較
pairs = sc.parallelize([("a", 1), ("b", 1), ("a", 1)] * 1_000_000)
pairs.groupByKey().mapValues(sum)    # 所有 value 都被 shuffle，reducer 端可能 OOM
pairs.reduceByKey(lambda x, y: x + y) # 先在 map 端局部加總，只 shuffle 部分和
# DataFrame 的 groupBy().agg() 會自動做 partial aggregation，不需手動選

Partition 數量調整

經驗法則：每個 partition 約 100–200MB（處理後的資料量），partition 數至少是總 core 數的 2–3 倍以避免閒置。太少 → 每個 task 太大、spill/OOM、平行度不足；太多 → 排程開銷、輸出大量小檔案。

Cache / Persist

因為 lazy evaluation，同一個 DataFrame 被多個 action 使用時，每次都會從頭重算。cache() / persist(level) 讓第一次計算後的結果保留在 executor 記憶體或磁碟。

• DataFrame 的 cache() 預設層級為 MEMORY_AND_DISK；RDD 的 cache() 預設為 MEMORY_ONLY。
• cache() 本身也是 lazy 的，要等下一個 action 才真的存。
• 只在「被重複使用 + 重算成本高」時才 cache；用完記得 unpersist()，否則佔住 executor 記憶體反而造成其他 stage spill。
• 需要切斷過長 lineage（例如迭代演算法）時用 checkpoint()，會寫到可靠儲存並截斷 DAG。

Broadcast variable 與 Accumulator

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("demo").getOrCreate()
sc = spark.sparkContext

# Broadcast variable：唯讀資料，每個 executor 只傳一次（而非每個 task 都序列化一份）
country_map = sc.broadcast({"TW": "Taiwan", "JP": "Japan", "SG": "Singapore"})
rdd = sc.parallelize(["TW", "JP", "XX"])
print(rdd.map(lambda c: country_map.value.get(c, "Unknown")).collect())

# DataFrame 中強制 broadcast join
small = spark.read.parquet("s3://bucket/dim_campaign/")
big = spark.read.parquet("s3://bucket/impressions/dt=2026-11-01/")
joined = big.join(F.broadcast(small), "campaign_id")

# Accumulator：task 只能累加、driver 才能讀；適合計數/除錯統計
bad_rows = sc.accumulator(0)
def parse(line):
    try:
        return [int(line)]
    except ValueError:
        bad_rows.add(1)
        return []
nums = sc.parallelize(["1", "x", "3"]).flatMap(parse)
nums.count()
print(bad_rows.value)   # 1

Catalyst、Tungsten 與 AQE

Catalyst optimizer

把 DataFrame/SQL 轉成邏輯計畫 → 解析（analysis，對照 catalog 解析欄位與型別）→ 邏輯最佳化（predicate pushdown、column pruning、constant folding、join reorder）→ 產生多個物理計畫並以成本模型選擇（例如選 broadcast hash join 還是 sort-merge join）→ code generation。用 df.explain(True) 可以看到各階段計畫。

Tungsten

執行層最佳化：以緊湊的二進位格式（UnsafeRow）管理記憶體、可用 off-heap 減少 GC 壓力、whole-stage code generation（把一個 stage 中的多個運算子融合成一個 Java 函式，避免虛擬函式呼叫與中間物件）、cache-friendly 的演算法。

AQE（Adaptive Query Execution）

Spark 3.0 引入、3.2 起預設開啟。在每個 shuffle stage 完成後，依實際的統計數據重新最佳化剩下的計畫：(1) 自動合併過小的 shuffle partition；(2) 發現某邊實際很小時把 sort-merge join 改成 broadcast join；(3) 拆分傾斜的 partition（skew join）。

spark.conf.set("spark.sql.adaptive.enabled", "true")                    # 3.2+ 預設已開
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(50 * 1024 * 1024))  # 調到 50MB
spark.conf.set("spark.sql.shuffle.partitions", "800")   # AQE 開啟時可設大一點，讓它往下合併

df.explain(mode="formatted")   # 看物理計畫：BroadcastHashJoin / SortMergeJoin / Exchange（= shuffle）

Spark 效能問題與 PySpark 實戰：上億筆 User Attributes 合併 高頻 深入

PySpark 範例：兩份 user attributes 合併，取每個欄位的最新非空值

這就是「上億筆怎麼 merge」的分散式標準答案。思路：把兩份資料 unionByName 成一張長表（每筆帶 source 與 updated_at），依 user_id 分組，用 window function 取最新的那一筆，或是逐欄位取最新的非空值。

from pyspark.sql import SparkSession, Window, functions as F

spark = (SparkSession.builder
         .appName("merge-user-attributes")
         .config("spark.sql.adaptive.enabled", "true")
         .getOrCreate())

crm = (spark.read.parquet("s3://lake/raw/crm_users/dt=2026-11-01/")
       .select(F.lower(F.trim("user_id")).alias("user_id"),           # key 正規化
               "email", "city", "tier",
               F.col("modified_time").cast("timestamp").alias("updated_at"))
       .withColumn("source", F.lit("crm"))
       .withColumn("priority", F.lit(2)))

app = (spark.read.parquet("s3://lake/raw/app_profiles/dt=2026-11-01/")
       .select(F.lower(F.trim("uid")).alias("user_id"),
               F.col("mail").alias("email"), "city",
               F.col("last_update").cast("timestamp").alias("updated_at"))
       .withColumn("tier", F.lit(None).cast("string"))                 # 補齊欄位
       .withColumn("source", F.lit("app_sdk"))
       .withColumn("priority", F.lit(1)))

unioned = (crm.unionByName(app)
           .filter(F.col("user_id").isNotNull() & (F.col("user_id") != "")))

# --- 做法 A：整筆記錄取最新（row-level latest wins）---
w_row = Window.partitionBy("user_id").orderBy(F.col("updated_at").desc_nulls_last(),
                                             F.col("priority").desc())
latest_row = (unioned
              .withColumn("rn", F.row_number().over(w_row))
              .filter("rn = 1")
              .drop("rn"))

# --- 做法 B：逐欄位取「最新的非空值」（field-level latest non-null）---
w_all = w_row.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
attrs = ["email", "city", "tier"]
field_level = (unioned
               .select("user_id",
                       *[F.first(c, ignorenulls=True).over(w_all).alias(c) for c in attrs],
                       F.max("updated_at").over(Window.partitionBy("user_id")).alias("updated_at"))
               .dropDuplicates(["user_id"]))

# --- 做法 C：用 groupBy + max_by（Spark 3.3+ 的 Python API）一次 shuffle 完成 ---
# 注意 max_by 不會略過 null，若要非空值優先，先把 null 欄位的排序 key 設成極小值
by_group = unioned.groupBy("user_id").agg(
    *[F.max_by(c, F.when(F.col(c).isNotNull(), F.col("updated_at"))).alias(c) for c in attrs],
    F.max("updated_at").alias("updated_at"))

(field_level
 .repartition(256, "user_id")                       # 控制輸出檔數與大小
 .write.mode("overwrite")
 .parquet("s3://lake/curated/user_profile/dt=2026-11-01/"))

每日增量合併（upsert）

每天全量重算數億筆不一定必要。若今天只有 500 萬人有變動，可以：昨天的 profile 快照 ⨝ 今天的增量 → 新快照。用 Delta Lake / Iceberg 時可以直接寫 MERGE INTO：

MERGE INTO curated.user_profile AS t
USING staging.user_profile_delta AS s
ON t.user_id = s.user_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET
  email      = COALESCE(s.email, t.email),          -- non-null wins
  city       = COALESCE(s.city,  t.city),
  updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT *;

常見效能問題與解法

檔案格式：CSV/JSON vs Parquet/ORC vs Avro 高頻 進階

在大數據場景中，選對檔案格式常常比調參數有用十倍。核心區別是資料在磁碟上的排列方式：行式（row-oriented，一筆記錄的所有欄位放在一起）或列式（column-oriented，同一欄位的所有值放在一起）。

Parquet 內部結構與兩大最佳化

• 檔案 → row group（一批列，常見設定約 128MB）→ column chunk（該 row group 中某一欄的資料）→ page（編碼與壓縮單位）。
• Footer 存放 schema 以及每個 row group、每個 column chunk 的統計（min、max、null count）。讀取端先讀 footer 決定要讀哪些部分。
• Column pruning：SELECT user_id, city 只讀這兩個 column chunk，其他欄位完全不碰磁碟。
• Predicate pushdown：WHERE event_ts >= '2026-11-01' 時，若某 row group 的 event_ts max 小於該值，整個 row group 跳過。資料若事先依過濾欄位排序或 clustering，min/max 範圍才會窄，跳過的效果才好。

import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.dataset as ds

table = pa.table({"user_id": ["u1", "u2", "u3"], "city": ["Taipei", "Tokyo", None],
                  "event_ts": pa.array([1730419200, 1730505600, 1730592000], pa.int64())})
pq.write_table(table, "users.parquet", compression="zstd", row_group_size=1_000_000)

# 只讀兩欄 + 過濾：pyarrow 會用 footer 統計做 row group 跳過
dataset = ds.dataset("users.parquet", format="parquet")
result = dataset.to_table(columns=["user_id", "city"],
                          filter=ds.field("event_ts") >= 1730500000)
print(result.to_pydict())
print(pq.ParquetFile("users.parquet").metadata.row_group(0).column(2).statistics)

壓縮演算法

Data Lake、Lakehouse、資料倉儲與 OLTP vs OLAP 進階

OLTP vs OLAP

Data Lake 與 Lakehouse

Data lake：把各種原始資料以開放格式（Parquet、JSON、Avro）放在便宜的物件儲存（S3、GCS），用 Spark / Trino / Hive 查詢。優點是便宜、彈性；缺點是原本沒有交易保證：寫到一半失敗留下殘檔、同時寫入互相覆蓋、無法有效率地 update/delete（GDPR 刪除請求就很痛苦）、schema 變更混亂（俗稱 data swamp）。

Lakehouse / Open table format 在物件儲存上加一層 metadata，提供 ACID 交易：

Apache Iceberg

以 metadata 檔 + manifest 追蹤每個 snapshot 包含哪些資料檔。支援 hidden partitioning（查詢者不需知道分區欄位怎麼算）、partition evolution、schema evolution、time travel、MERGE INTO。引擎中立（Spark、Flink、Trino、Snowflake 等都支援），近年採用度很高。

Delta Lake

由 Databricks 開源，以 _delta_log 交易日誌（JSON + 定期 checkpoint）記錄每次 commit。與 Spark 整合最緊密，支援 time travel、OPTIMIZE（合併小檔案）、Z-ordering。

Apache Hudi

源自 Uber，重點在高頻 upsert 與增量讀取。兩種表型：Copy-on-Write（寫入時重寫檔案，讀快寫慢）與 Merge-on-Read（寫入 log 檔，讀取時合併，寫快讀慢）。

雲端資料倉儲與 OLAP 引擎

批次 vs 串流、Lambda/Kappa、Flink 與 Structured Streaming 進階 深入

批次 vs 串流

Lambda vs Kappa 架構

實務上很多公司是「混合」：即時指標走串流（例如即時花費、pacing），隔天用批次重算出精確的帳務數字並覆寫。這對廣告業特別常見：即時數字用於決策，批次數字用於計費與對帳。

Flink 核心概念

Event time vs Processing time

Event time 是事件實際發生的時間（寫在事件裡，例如手機 SDK 的時間戳）；processing time 是運算元處理它時的系統時鐘。以 processing time 計算的「每分鐘點擊數」會受網路延遲、重播、積壓影響而失真；正確的業務統計應用 event time。

Watermark

串流中的一個特殊標記 W(t)，宣告「event time ≤ t 的事件應該都到了」。當 watermark 超過某個 window 的結束時間，該 window 才觸發計算。常見策略：watermark = 目前看到的最大 event time − 允許亂序時間（bounded out-of-orderness）。Watermark 太保守 → 延遲高；太激進 → 更多遲到資料被丟棄或需另外處理（allowed lateness、side output）。

Window 類型

Tumbling（固定大小、不重疊：每 1 分鐘一個窗）、Sliding（固定大小、有重疊：長度 10 分鐘、每 1 分鐘滑動，一個事件屬於多個窗）、Session（依活動間隔切分：超過 30 分鐘沒事件就結束一個 session，窗長度不固定）、以及 Global window 搭配自訂 trigger。

State

運算元需要記住的資料（例如每個 user 目前的合併屬性、每個 campaign 的累計花費）。Keyed state 依 key 分區，與資料一起 partition。State backend：HashMap（在 JVM heap，快但受記憶體限制）或 RocksDB（嵌入式 LSM 存在本地磁碟，可支撐遠大於記憶體的狀態，支援增量 checkpoint）。

Checkpoint

Flink 定期在資料流中插入 checkpoint barrier，各運算元收到 barrier 時把狀態快照到持久儲存（S3/HDFS），基於 Chandy-Lamport 分散式快照的變形。失敗時從最近的 checkpoint 還原狀態，並把 source（例如 Kafka offset）倒回對應位置重播。Savepoint 是手動觸發、用於升級/遷移的快照。

Exactly-once

Flink 內部狀態的 exactly-once 由 checkpoint 保證。端到端 exactly-once 還需要：可重播的 source（Kafka）+ 交易式或冪等的 sink（例如 Kafka transactional producer 搭配兩階段提交：checkpoint 完成時才 commit 交易；或寫入時以唯一 key upsert）。下游讀 Kafka 時要設 isolation.level=read_committed 才看不到未提交的資料。

即時版本的 user attributes 合併

若面試官要求「使用者屬性一更新，幾秒內就要反映到線上」，概念上是一個以 user_id 為 key 的有狀態串流：每來一個屬性事件，就讀出該 user 目前的狀態、套用衝突規則、寫回狀態並輸出到線上 KV store。以下用純 Python 示意其邏輯（實務上由 Flink keyed state 或 Kafka Streams KTable 提供分區、容錯與重播）：

from dataclasses import dataclass, field

@dataclass
class ProfileState:
    fields: dict[str, tuple[object, int]] = field(default_factory=dict)  # name -> (value, event_ts)

def process(state: dict[str, ProfileState], event: dict, sink) -> None:
    """event: {"user_id": ..., "event_ts": ..., "attrs": {...}}；可亂序到達。"""
    st = state.setdefault(event["user_id"], ProfileState())
    changed = False
    for name, value in event["attrs"].items():
        if value is None:
            continue                                   # non-null wins
        old = st.fields.get(name)
        if old is None or event["event_ts"] >= old[1]: # 依 event time，而非到達順序
            st.fields[name] = (value, event["event_ts"])
            changed = True
    if changed:
        # 對 sink 做 upsert（以 user_id 為主鍵）→ 重播同一事件也不會產生重複：冪等
        sink.upsert(event["user_id"], {k: v for k, (v, _) in st.fields.items()})

Spark Structured Streaming vs Flink

from pyspark.sql import functions as F

clicks = (spark.readStream.format("kafka")
          .option("kafka.bootstrap.servers", "kafka:9092")
          .option("subscribe", "ad_clicks")
          .load()
          .select(F.from_json(F.col("value").cast("string"),
                              "campaign_id STRING, user_id STRING, event_ts TIMESTAMP").alias("e"))
          .select("e.*"))

per_minute = (clicks
              .withWatermark("event_ts", "2 minutes")                  # 允許 2 分鐘亂序
              .groupBy(F.window("event_ts", "1 minute"), "campaign_id") # tumbling window
              .agg(F.count("*").alias("clicks"),
                   F.approx_count_distinct("user_id").alias("uu")))     # HyperLogLog 近似 UU

(per_minute.writeStream
 .outputMode("update")
 .option("checkpointLocation", "s3://bucket/checkpoints/clicks_per_minute/")
 .format("console")
 .trigger(processingTime="10 seconds")
 .start())

資料管線編排、冪等重跑與資料品質 進階

當 merge 要「每天跑」，問題就從演算法變成工程可靠性：失敗怎麼重試？上游資料晚到怎麼辦？發現 bug 要重算過去 30 天怎麼辦？資料壞掉誰會知道？

Airflow DAG 範例

Airflow 用 Python 定義 DAG（有向無環圖），每個節點是一個 task，排程器依時間與相依關係觸發。關鍵概念：每次執行對應一個邏輯日期（logical date，舊稱 execution_date），task 應該只處理「該邏輯日期」的資料，而不是「現在」的資料。以下以 Airflow 2.x 語法為例：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor

default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=10),
    "retry_exponential_backoff": True,
}

with DAG(
    dag_id="daily_user_profile_merge",
    start_date=datetime(2026, 1, 1),
    schedule="0 2 * * *",          # 每天 02:00（依 DAG 時區設定）
    catchup=False,                 # 新部署時不自動補跑歷史；需要時手動 backfill
    max_active_runs=1,
    default_args=default_args,
    tags=["profile"],
) as dag:
    wait_crm = ExternalTaskSensor(
        task_id="wait_crm_export",
        external_dag_id="crm_export",
        external_task_id="done",
        mode="reschedule",             # 等待時釋放 worker slot
        timeout=6 * 3600,
    )
    merge = BashOperator(
        task_id="spark_merge",
        # {{ ds }} 是邏輯日期（YYYY-MM-DD）：同一個 ds 無論重跑幾次都處理同一天的資料
        bash_command="spark-submit merge_profiles.py --dt {{ ds }}",
    )
    validate = BashOperator(
        task_id="validate",
        bash_command="python validate_profile.py --dt {{ ds }}",
    )
    publish = BashOperator(
        task_id="publish_to_online_store",
        bash_command="python publish.py --dt {{ ds }}",
    )
    wait_crm >> merge >> validate >> publish

冪等（idempotent）與 backfill

冪等的意思是：同一個邏輯日期的 job 跑一次與跑十次，結果完全相同。這是能安全重試與 backfill 的前提。

• 輸入固定：依 {{ ds }} 讀取對應分區，不要用 datetime.now()，也不要讀「最新」的資料。
• 輸出覆寫而非附加：INSERT OVERWRITE ... PARTITION (dt=...)、Spark mode("overwrite") 搭配 spark.sql.sources.partitionOverwriteMode=dynamic（只覆寫有寫到的分區，避免整張表被清空）。用 append 重跑就會產生重複資料。
• 原子發佈：先寫到暫存位置，驗證通過後再切換（rename、更新指標、table format 的原子 commit），讀取端永遠看不到寫一半的資料。
• 確定性：衝突規則要有 deterministic tie-break（見第 2 節），避免 row_number() 在平手時每次結果不同。
• Backfill：修了 bug 要重算 10 月：airflow dags backfill -s 2026-10-01 -e 2026-10-31 daily_user_profile_merge（Airflow 2.x CLI）。若每天的結果依賴前一天（增量快照），backfill 必須依序執行（max_active_runs=1 或 depends_on_past=True）。

資料品質檢查

from pyspark.sql import functions as F

def validate(df, dt: str, expected_min: int) -> None:
    stats = df.agg(
        F.count("*").alias("rows"),
        F.countDistinct("user_id").alias("uniq"),
        F.sum(F.col("email").isNull().cast("int")).alias("null_email"),
    ).first()
    errors = []
    if stats["rows"] < expected_min:
        errors.append(f"row count {stats['rows']} < {expected_min}")
    if stats["rows"] != stats["uniq"]:
        errors.append(f"duplicate user_id: {stats['rows'] - stats['uniq']}")
    if stats["rows"] and stats["null_email"] / stats["rows"] > 0.5:
        errors.append("null email ratio > 50%")
    if errors:
        raise ValueError(f"[{dt}] data quality failed: {errors}")   # 讓 Airflow task 失敗

CUDA / GPU 平行運算：一句話概念與何時用

一句話：GPU 有數千個簡單核心，以 SIMT（同一指令、多執行緒）方式對大量資料做同樣的數值運算；CUDA 是 NVIDIA 讓你撰寫這類 kernel 的平台。

• 適合：資料平行、運算密集、分支少的工作 —— 深度學習訓練與推論、矩陣運算、向量相似度搜尋、大規模數值模擬；在資料處理上也有 RAPIDS（cuDF、Spark RAPIDS Accelerator）把 DataFrame 運算搬到 GPU。
• 不適合：I/O-bound 的工作（瓶頸在讀檔/網路，GPU 幫不上）、分支很多的邏輯、資料量小（CPU 與 GPU 之間透過 PCIe 搬資料的成本可能大於運算省下的時間）、需要大量隨機存取的字串處理。
• 面試回答：「上億筆 merge 本質上是 I/O 與 shuffle 密集，GPU 通常不是第一選擇；但如果後續要訓練 CTR 模型或做 embedding 相似度計算，GPU 就很關鍵。」

Appier 情境：跨渠道使用者資料整合與 Identity Resolution 深入

「merge user attributes from multiple data sources」在 AdTech / MarTech 中不只是 join：同一個人可能在不同渠道有不同的 ID —— 網站 cookie、App 的裝置 ID（IDFA/GAID 或 app 自有 ID）、登入後的會員 ID、email、手機號碼、LINE ID。要先回答「哪些 ID 屬於同一個人」，才能合併屬性。這就是 identity resolution（身分解析/ID 打通）。

Deterministic matching（確定性比對）

靠明確的共同識別碼連結：同一個登入帳號在網站與 App 都出現、同一個 hashed email。準確但覆蓋率受限於登入率。

Probabilistic matching（機率性比對）

靠訊號推測：相同 IP + 相似時間 + 裝置特徵。覆蓋率高但有誤判，且受隱私法規與平台政策嚴格限制（例如裝置指紋在多數平台政策下不被允許），實務上要非常謹慎。

Identity graph

節點是各種 ID，邊是「曾經一起出現」的證據。同一個人 = 圖中的一個連通分量（connected component）。

單機版：Union-Find 求連通分量

class UnionFind:
    def __init__(self) -> None:
        self.parent: dict[str, str] = {}
        self.rank: dict[str, int] = {}

    def find(self, x: str) -> str:
        self.parent.setdefault(x, x)
        self.rank.setdefault(x, 0)
        root = x
        while self.parent[root] != root:         # 迭代版，避免遞迴深度問題
            root = self.parent[root]
        while self.parent[x] != root:            # path compression
            self.parent[x], x = root, self.parent[x]
        return root

    def union(self, a: str, b: str) -> None:
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return
        if self.rank[ra] < self.rank[rb]:       # union by rank
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1

# 每一列是一次「同時出現」的證據
links = [
    ("cookie:c1", "member:m100"),     # 網站登入
    ("device:d9", "member:m100"),     # App 登入
    ("device:d9", "email:h(a@x.com)"),
    ("cookie:c7", "email:h(b@y.com)"),
]
uf = UnionFind()
for a, b in links:
    uf.union(a, b)

groups: dict[str, list[str]] = {}
for node in list(uf.parent):
    groups.setdefault(uf.find(node), []).append(node)
print(list(groups.values()))
# [['cookie:c1', 'member:m100', 'device:d9', 'email:h(a@x.com)'], ['cookie:c7', 'email:h(b@y.com)']]

近乎 O(α(n)) 的均攤複雜度（α 為反 Ackermann 函數，實務上視為常數）。得到每個 ID 對應的「統一 ID（root）」後，就能把各渠道的屬性依統一 ID 用前面的衝突規則合併。

上億個節點：分散式連通分量

• 迭代式 label propagation：每個節點的 label 初始為自己，每輪取鄰居中最小的 label，直到不再變化。每輪是一次 Spark join + groupBy，輪數約為圖的直徑（通常不大，但遇到長鏈會很多輪）。
• GraphFrames 的 connectedComponents()（Spark 上的圖運算套件）實作了更有效率的演算法，並需要設定 checkpoint 目錄截斷 lineage。
• 防止「超級節點」：公共 Wi-Fi 的 IP、共用平板、測試帳號、錯誤的預設值（例如所有人的 email 都是 test@test.com）會把數百萬人連成一個巨大分量。必須設定規則：某個 ID 連到超過 N 個其他 ID 就視為不可信、不參與合併。這是真實系統最常出事的地方，面試提到會很加分。

完整面試回答範本：逐層追問對答劇本 高頻

以下是把本章串起來的模擬對話。重點不是背台詞，而是學會每一層的回答結構：釐清 → 方案 → 複雜度 → trade-off → 主動提下一步。

第一層：「兩份資料怎麼 merge？」

第二層：「資料量增加到上億筆怎麼辦？」

第三層：「機器記憶體只有 8GB，而且兩份都很大？」

第四層：「這要每天跑，而且單機要跑好幾個小時？」

第五層：「如果要即時，屬性一變就要生效？」

可能的延伸追問與一句話回答

面試自測題