一. 前言：Join 沒報錯，營收卻變多了 #
嗨，我是拍拍君。今天的任務很單純：替訂單補上客戶名稱。 五筆訂單進去，七筆資料出來，程式還笑咪咪地正常結束。 不是業績突然變好，是右表同一個客戶出現兩次。
Join 的成功條件，不是「有結果」，而是結果仍符合資料關係。 這次不列一整頁 Join 類型，而是替一條補資料管線訂契約： 一列仍是一筆訂單、金額不變、客戶最多配一筆、漏配原因要能對帳。
如果剛接觸 expression，先看 Polars 基礎。 本篇比 Pandas 資料清理 的簡短 merge 範例更深入， 重點是複合鍵、配對基數與來源辨識，不是再做一次清洗流程。
二. 安裝：固定版本，也固定「一列代表什麼」 #
以下小型範例實測 Python 3.12、Polars 2.0.0：
uv init pypy-join-contracts
cd pypy-join-contracts
uv add 'polars==2.0.0'
uv add --dev pytest
第三節存成 fixtures.py；第八節存成 join_contract.py。
第四至七節、九節的 Python 片段依序放入 demo.py，最後再一起執行。
先寫契約，才知道要檢查哪一邊：
| 對象 | 粒度／規則 |
|---|---|
| 訂單 facts | 一列一個 order_id，ID 不可空且唯一 |
| 客戶 dimension | 一列一組 (shop, customer_id)，鍵不可空且唯一 |
| 補完資料 | 每筆訂單恰好保留一次，原始欄位不可改 |
| 缺少客戶鍵 | 保留訂單，標成 null_key，不能亂補客戶 |
| 鍵存在但查不到 | 保留訂單，標成 orphan_key |
後兩種資料可以留在診斷輸出，但預設不能通過正式報表 gate。 這是本篇選擇的政策；你的業務若允許訪客訂單，應明確修改規則。
三. 測試資料：同一個 c1，不一定是同一個人 #
客戶 ID 只在各店內唯一，因此不能單用 customer_id。
把以下存成 fixtures.py；schema 常數定義在第八節：
import polars as pl
from join_contract import FACT_SCHEMA, DIM_SCHEMA
orders = pl.DataFrame({
"order_id": [101, 102, 103, 104, 105],
"shop": ["tw", "tw", "jp", "tw", "tw"],
"customer_id": ["c1", "c1", "c1", None, "c404"],
"amount_cents": [100, 200, 300, 400, 500],
}, schema=FACT_SCHEMA)
customers = pl.DataFrame({
"shop": ["tw", "jp", "tw"],
"customer_id": ["c1", "c1", "c2"],
"customer_name": ["拍拍醬", None, "chatPTT"],
}, schema=DIM_SCHEMA)
金額是同一幣別的整數分，只用來核對是否重算，不做匯率或捨入。
jp/c1 有客戶紀錄，但名稱尚未補齊；tw/c2 目前沒有訂單。
104 缺少客戶 ID，105 則帶著查不到的 c404。
這些不是同一種錯誤，別通通叫「有 null」。
四. 先抓重複鍵：不是只找整列一模一樣 #
先製造一個名稱不同、但鍵相同的客戶版本：
import polars as pl
from fixtures import orders, customers
from join_contract import KEYS, duplicate_keys, null_key
dirty = pl.concat([
customers,
pl.DataFrame({"shop": ["tw"], "customer_id": ["c1"],
"customer_name": ["拍拍君"]}, schema=customers.schema),
])
assert duplicate_keys(dirty).to_dicts() == [
{"shop": "tw", "customer_id": "c1", "key_count": 2}
]
unsafe = orders.join(dirty, on=KEYS, how="left")
assert unsafe.height == 7
assert unsafe["amount_cents"].sum() == 1800 # 原本只有 1500
同一個非空鍵，左邊有 L 筆、右邊有 R 筆，就會產生 L × R 個配對。
這裡 tw/c1 的兩筆訂單各複製一次，所以多出兩列、多算 300 分。
其他漏配訂單仍保留一列，不能拿「左表五筆」推論 left join 永遠五筆。
is_duplicated() 檢查整列時，看不出兩個不同名稱其實搶同一個鍵。
應按預定的 Join key 分組計數，再回查那些版本的來源。
若只按 customer_id 分組，又會把不同店的合法 c1 誤判成重複。
不要直接 unique(keep="first") 把錯誤壓掉。
同一個客戶有多個歷史版本時，需要有效期間或已核准的版本選擇規則；
任選第一列，不等於解決主資料衝突。
五. Cardinality：把預期關係交給 validate #
本例左表可以重複客戶，右表不可以，所以是 m:1：
safe = orders.join(
customers, on=KEYS, how="left", validate="m:1",
nulls_equal=False, maintain_order="left",
)
assert safe.height == 5
try:
orders.join(dirty, on=KEYS, how="left", validate="m:1")
except pl.exceptions.ComputeError:
pass # 預期失敗：右表的複合鍵不唯一
else:
raise AssertionError("重複客戶鍵竟然通過")
依 官方 Join API，
validate 的左右，是依本次呼叫的左表與右表，不是你心中的主表：
| validate | 檢查哪邊的 Join key 唯一 |
|---|---|
1:1 |
兩邊 |
1:m |
左邊 |
m:1 |
右邊 |
m:m |
不做唯一性檢查，也是預設值 |
這組訂單若改用 1:1 會失敗，因為 101、102 都屬於 tw/c1。
訂單 ID 唯一，不代表訂單上的客戶鍵也唯一。
validate="m:1" 更不保證每筆訂單都有客戶，它只限制配對的基數。
六. Null：不能拿「未知」當成同一個客戶 #
本篇顯式選 nulls_equal=False，不讓 Null key 彼此匹配。
先用另一組最小資料觀察差異，不修改正式客戶表：
left = pl.DataFrame({"k": [None, None]}, schema={"k": pl.String})
right = pl.DataFrame({"k": [None, None], "v": ["A", "B"]},
schema={"k": pl.String, "v": pl.String})
assert left.join(right, on="k", how="left", nulls_equal=False).height == 2
assert left.join(right, on="k", how="left", nulls_equal=True).height == 4
允許 Null 相等會讓兩筆「不知道是誰」配到兩筆「也不知道是誰」。 不是所有缺值都代表同一個實體，因此不能把這個開關當作修復缺漏。 複合鍵只要任一欄是 null，本篇就判為缺鍵；右表則直接拒絕缺鍵。
注意：validate 不是 NOT NULL 約束。
實測 nulls_equal=False 時，右表的重複 Null 不一定觸發 m:1 錯誤；
所以不可空政策要獨立檢查，而不是等 Join 替你報錯。
空字串也不是 null；是否允許空白、是否大小寫合併，應在入口明確決定。
正規化必須在唯一性檢查前完成，並保留原始值以便診斷碰撞。
七. Anti Join：漏的是訂單，還是沒被使用的客戶？ #
只看結果的 customer_name.is_null() 不可靠：
103 其實配對成功，只是客戶名稱為 null。
先把缺鍵和有鍵但不存在的訂單拆開：
missing_key = orders.filter(null_key())
present_key = orders.filter(~null_key())
orphans = present_key.join(customers, on=KEYS, how="anti", nulls_equal=False)
matched = present_key.join(customers, on=KEYS, how="semi", nulls_equal=False)
unused = customers.join(present_key, on=KEYS, how="anti", nulls_equal=False)
assert missing_key["order_id"].to_list() == [104]
assert orphans["order_id"].to_list() == [105]
assert matched.sort("order_id")["order_id"].to_list() == [101, 102, 103]
assert unused.select(KEYS).rows() == [("tw", "c2")]
assert orders.height == missing_key.height + orphans.height + matched.height
Anti join 保留左側沒有配對的列，不會把右側欄位補過來。 換左右之後，問題也跟著改變：未匹配訂單是參照完整性問題； 沒有被訂單使用的客戶，則可能完全合法，不必自動刪掉。
計算 orphan rate 時要說清楚分母：本例是有鍵訂單的 1/4， 若報全體訂單的未匹配率，則是缺鍵加 orphan 的 2/5。 不要換了分母，卻仍把兩個數字叫成同一項指標。
八. 完整管線：配對標記、粒度不變量與預設拒收 #
把下面存成 join_contract.py。
此函式刻意只接受兩份指定 schema，額外欄位也拒絕，避免 suffix 或內部欄撞名。
右表加一個永遠為 True 的標記，再 Join；這才能區分「配不到」和「屬性空值」。
import polars as pl
KEYS = ["shop", "customer_id"]
FACT_SCHEMA = {"order_id": pl.Int64, "shop": pl.String,
"customer_id": pl.String, "amount_cents": pl.Int64}
DIM_SCHEMA = {"shop": pl.String, "customer_id": pl.String,
"customer_name": pl.String}
class ContractError(ValueError):
pass
def require(ok, code):
if not ok:
raise ContractError(code)
def null_key():
return pl.any_horizontal([pl.col(k).is_null() for k in KEYS])
def duplicate_keys(frame):
return (frame.group_by(KEYS).len(name="key_count")
.filter(pl.col("key_count") > 1).sort(KEYS))
def enrich_orders(facts, dimension, *, allow_unmatched=False):
require(dict(facts.schema) == FACT_SCHEMA, "fact_schema")
require(dict(dimension.schema) == DIM_SCHEMA, "dimension_schema")
require(facts["order_id"].null_count() == 0, "null_order_id")
require(facts["order_id"].n_unique() == facts.height, "duplicate_order_id")
require(dimension.filter(null_key()).is_empty(), "null_dimension_key")
require(duplicate_keys(dimension).is_empty(), "duplicate_dimension_key")
joined = facts.join(
dimension.with_columns(pl.lit(True).alias("_matched")),
on=KEYS, how="left", validate="m:1",
nulls_equal=False, maintain_order="left",
)
result = joined.with_columns(
pl.when(null_key()).then(pl.lit("null_key"))
.when(pl.col("_matched").fill_null(False)).then(pl.lit("matched"))
.otherwise(pl.lit("orphan_key")).alias("join_status")
).select(*facts.columns, "customer_name", "join_status")
require(result.height == facts.height, "row_count_changed")
require(result.select(facts.columns).equals(facts), "fact_columns_changed")
counts = dict(result.group_by("join_status").len().iter_rows())
report = {k: counts.get(k, 0) for k in ("matched", "null_key", "orphan_key")}
require(sum(report.values()) == facts.height, "reconciliation_failed")
if not allow_unmatched:
require(report["null_key"] + report["orphan_key"] == 0, "unmatched_key")
return result, report
require 用例外實作正式 gate；不使用可能被 python -O 移除的 assert。
對 facts 的整份投影比對，比只檢查金額總和更強：列數、ID、順序、值都沒改。
上面是小型 eager enrichment，不是任意資料量都能裝進 RAM 的通用平台。
它也不驗證金額正負、名稱長度或每個店是否合法；那些仍由上游契約負責。
九. 診斷可以保留，正式通過要有政策 #
回到 demo.py，明確允許診斷輸出：
from join_contract import enrich_orders
result, report = enrich_orders(orders, customers, allow_unmatched=True)
assert report == {"matched": 3, "null_key": 1, "orphan_key": 1}
assert result["join_status"].to_list() == [
"matched", "matched", "matched", "null_key", "orphan_key"
]
assert result["amount_cents"].sum() == orders["amount_cents"].sum() == 1500
assert result.filter(pl.col("order_id") == 103)["customer_name"].item() is None
103 的空名稱沒有被誤認成漏配，104、105 也沒有從結果消失。
正式呼叫省略 allow_unmatched=True，這份 fixture 就會因 unmatched_key 失敗。
先對帳、修來源或明確核准例外，再發布報表；別悄悄改成 inner join 丟掉兩筆。
若需要隔離清單，從 result 按 join_status 輸出，另存 batch ID 與來源版本。
不要在公開 log 印出整份客戶資料。
十. 測試：壞資料必須真的被擋住 #
以下存成 test_join_contract.py，不只檢查「成功回傳一張表」：
import polars as pl
import pytest
from fixtures import orders, customers
from join_contract import ContractError, enrich_orders
def test_status_and_grain():
result, report = enrich_orders(orders, customers, allow_unmatched=True)
assert result.select(orders.columns).equals(orders)
assert report == {"matched": 3, "null_key": 1, "orphan_key": 1}
assert result["join_status"][2] == "matched" # 名稱 null 仍有配對
def test_duplicate_dimension_is_not_silently_deduplicated():
dirty = pl.concat([customers, customers.head(1)])
with pytest.raises(ContractError, match="^duplicate_dimension_key$"):
enrich_orders(orders, dirty, allow_unmatched=True)
def test_publication_gate_rejects_unmatched():
with pytest.raises(ContractError, match="^unmatched_key$"):
enrich_orders(orders, customers)
uv run python demo.py
uv run pytest -q
正式 regression suite 再加：左右空表、Null 複合鍵、重複訂單 ID、dtype 漂移， 以及「兩店使用相同客戶 ID」不能串錯人的案例。 Join 的基數驗證、缺值政策、完整性對帳，應該各有獨立的失敗案例。
十一. 換成 Lazy／大型資料時，不要把保證弄丟 #
本篇沒有示範 collect(engine="streaming")，因為重點是契約，不是速度。
目前官方 Join 文件仍註明 validation 的 streaming 支援限制；
不能把引擎選項換掉，就假設同樣的檢查一定被執行。
大型管線可以把唯一鍵檢查、anti join 對帳與正式 enrichment 拆成工作， 但它們必須讀同一份不可變來源快照，否則會發生「檢查完資料卻換了」的競態。 全量 uniqueness 檢查不是抽樣；在不同分區各自唯一，也不代表全域唯一。 輸出先進 staging，通過對帳再提升版本，避免下游讀到一半完成的結果。
同樣別依賴碰巧看到的 Join 順序：本例顯式 maintain_order="left"。
若順序不是契約的一部分，測試可以按穩定 ID 排序後再比；
若輸出刻意改為一筆訂單多個商品，就應改寫粒度與不變量，而不是關掉錯誤。
查詢計畫與資源量測請接著看 LazyFrame 最佳化。
結語：每一次 Join，都要回答「多了誰、少了誰」 #
拍拍君這次最想留下的不是一個 validate 參數，而是一組能對帳的關係：
鍵在哪個範圍唯一、最多可以配幾筆、缺鍵與漏配如何區分，最後一列仍代表什麼。
先診斷重複，再限制 cardinality；用 anti join 看漏配，用來源標記辨識空屬性。
當列數、原始事實與配對狀態都可驗證，漂亮的表格才比較接近可信的答案。