Polars Nested JSON 實戰：Struct、List、NDJSON 與 Schema Drift

Python 學習 - 本文屬於一個選集。

§ 140: 本文

一. 前言：API 回來的不是一張表
#

嗨，我是拍拍君。今天收到一份訂單匯出：每筆訂單有客戶物件，客戶裡有國家，訂單裡還有商品清單。把 JSON 讀進來不難，難的是「一列到底代表什麼」。

展開商品後，一筆訂單變成兩列；直接數列數，就多了一筆訂單。有些商品清單是 []，有些是 null，你又該保留哪種？今天不比誰快，而是把巢狀資料轉成粒度明確、契約可檢查的表。

如果還不熟 expression，先看 Polars 基礎；查詢計畫與 streaming 則留給 LazyFrame 最佳化。本篇的工作順序是：來源紀錄 → 契約檢查 → 巢狀欄位操作 → 商品明細 → QA。

二. 安裝：先固定可重現版本
#

本文實測 Polars 2.0.0，建立獨立專案：

uv init nested-orders
cd nested-orders
uv add 'polars==2.0.0'
uv add --dev pytest
uv run python -c "import polars as pl; print(pl.__version__)"

以下 Python 片段按順序放進 demo.py，第十節另存 intake.py。教學資料很小，所以用 collect() 看答案；正式大檔不必每一步 materialize。範例金額一律為 TWD 的整數分，避免浮點數累積誤差。

三. 資料入口：NDJSON 不是漂亮排版的 JSON
#

NDJSON 是一行一筆完整紀錄，不是跨好幾行的 JSON array。格式細節可參考 Python JSON，這裡只建立測試資料：

import json
from pathlib import Path
import polars as pl
rows = [
    {"order_id": "P001", "customer": {"name": "拍拍君", "country": "TW"},
     "items": [{"sku": "book", "qty": 2, "unit_cents": 12000},
               {"sku": "pen", "qty": 1, "unit_cents": 3000}],
     "tags": [" NEW ", "gift"]},
    {"order_id": "P002", "customer": {"name": "拍拍醬"},
     "items": [], "tags": []},
    {"order_id": "P003", "customer": {"name": "chatPTT", "country": "TW"},
     "items": None, "tags": ["review"]},
]
source = Path("orders.ndjson")
source.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n"
                          for r in rows), encoding="utf-8")

三筆資料刻意包含「有商品」「確認沒有商品」「商品未知」。後兩者不能因為總額都看起來像零，就被當成同一種狀態。

小型資料用 read_ndjson() 得到 DataFrame；可組合管線用 scan_ndjson() 得到 LazyFrame。不要把 read_ndjson(...).lazy() 當成延後檔案讀取。

四. 明確 Schema：Struct 裝欄位，List 裝多個元素
#

這份資料的型別不是「一欄 object」，而是：

欄位	型別	粒度
`order_id`	String	一筆訂單的識別碼
`customer`	Struct	每筆訂單的一份客戶快照
`items`	List[Struct]	每筆訂單的多筆商品
`tags`	List[String]	每筆訂單的多個標籤

定義完整巢狀型別：

ITEM = pl.Struct({"sku": pl.String, "qty": pl.Int64,
                  "unit_cents": pl.Int64})
SCHEMA = {
    "order_id": pl.String,
    "customer": pl.Struct({"name": pl.String, "country": pl.String}),
    "items": pl.List(ITEM),
    "tags": pl.List(pl.String),
}
orders = pl.scan_ndjson(source, schema=SCHEMA)
print(orders.collect_schema())

官方 scan_ndjson 支援完整 schema 與局部 schema_overrides。沒有契約時，推論只能依取樣看到的內容猜型別；增加 infer_schema_length 能看到更多樣本，不能取代正式規格。

特別是某欄前幾百筆都為 null，後面才出現物件時，很容易猜錯。此外，dtype 並不代表欄位必填、字串非空或數量必須大於零。 Schema 是載入型別，不是完整的業務 validator。

五. Struct 欄位：先選需要的，不急著全部攤平
#

只需要國家時，用 struct.field()：

customers = orders.select(
    "order_id",
    pl.col("customer").struct.field("name").alias("customer_name"),
    pl.col("customer").struct.field("country").alias("country"),
)
print(customers.collect())

P002 的 country 沒有出現在來源物件裡，依指定型別載入為 null。但「欄位不存在」與「來源明確填 null」若需要分開追蹤，就要在入口保存 presence 資訊，不能指望 DataFrame 事後還原。

想一次拆出 Struct 全部欄位，可以 unnest()。先重新命名內部欄位，再拆，避免撞到外層同名的 name：

flat_customer = (
    orders.with_columns(
        pl.col("customer").struct.rename_fields(
            ["customer_name", "customer_country"]
        )
    ).unnest("customer")
)
assert flat_customer.collect().height == 3

unnest 改的是欄，不把一列變多列。欄位順序和數量要對應 schema；正式系統可改採逐欄 alias，較不依賴位置。

六. List 內部運算：不是每個需求都要 explode
#

只是清理標籤，不需要拆成多列：

normalized = orders.with_columns(
    pl.col("tags").list.eval(
        pl.element().str.strip_chars().str.to_lowercase()
    ).alias("tags_clean"),
    pl.col("items").list.len().alias("item_count"),
)
assert normalized.collect()["tags_clean"][0].to_list() == ["new", "gift"]

pl.element() 指目前 List 裡的元素，不是整張表的當前列。如果元素本身是 Struct，也能繼續 struct.field()：

summaries = orders.with_columns(
    pl.col("items").list.eval(
        pl.element().struct.field("qty")
        * pl.element().struct.field("unit_cents")
    ).list.sum().alias("total_cents")
)
summary = summaries.collect()
assert summary["total_cents"].to_list() == [27000, 0, None]

此處空 List 的 sum 是 0，null List 保持 null。這符合範例契約：已知沒有商品的訂單為零，未知商品不能憑空補零。要引用外層欄位參與商品計算時，通常先展開再用普通 expression，比混用 scope 清楚。

七. explode + unnest：一列訂單變成一列商品
#

商品銷售分析需要明細粒度。先排除空與未知清單，再拆兩層：

def detail_query(frame: pl.LazyFrame) -> pl.LazyFrame:
    return (
        frame.filter(pl.col("items").list.len().fill_null(0) > 0)
        .select("order_id", "items")
        .explode("items", empty_as_null=False, keep_nulls=False)
        .unnest("items")
        .with_columns(
            (pl.col("qty") * pl.col("unit_cents")).alias("line_cents")
        )
    )
details = detail_query(orders).collect()
assert details.height == 2
assert details["order_id"].to_list() == ["P001", "P001"]
assert details["line_cents"].sum() == 27000

explode 把 List 變成多列，unnest 再把每列 Struct 拆成欄。因此現在 pl.len() 是明細列數，不是訂單數。別在這張表直接加總原本的訂單總額，否則同一訂單會被重算。每筆金額雖已設界限，大批次聚合仍須監看 Int64 溢位風險。

兩個無關的 List 不要連續 explode；那會產生組合，可能是意外的笛卡兒積。平行陣列要先驗證每列長度一致，再同時展開；來源若可調整，直接提供 List[Struct] 通常更能保住配對關係。

八. 空清單與 null：先寫政策，再呼叫 API
#

本文顯式設定 explode 旗標，不依賴不同版本的預設值。需要保留所有訂單位置時，可以保留占位列：

placeholders = (
    orders.select("order_id", "items")
    .explode("items", empty_as_null=True, keep_nulls=True)
    .collect()
)
assert placeholders.height == 4  # 兩筆商品 + 空清單 + 未知清單

這張表後兩列的商品都是 null，不能把它們算成售出商品。若之後要辨認空與未知，請在 explode 前保存 items.is_null() 和 list.len()。

需求	建議
只分析已知商品	篩掉空/null List，再展開
對帳所有原始訂單	保留訂單主表，明細另表
畫面顯示沒有商品的訂單	保存狀態欄，再保留占位列

不要為了方便報表，偷偷把未知資料變成已知零。

九. Schema Drift：型別能載入，不代表變更已核准
#

來源今天新增 discount，明天把 qty 改成字串，後天 customer 變清單。這些都是 drift，但不一定該用同一種策略：

新增非必要欄位：先記錄、評估，再升版契約；
必要欄位遺失：隔離該紀錄，不用 null 假裝成功；
字串變數字：只有契約明確允許時才正規化；
物件變清單：視為結構變更，不猜哪個元素才正確。

ignore_errors=True 不是壞紀錄收容所。官方說明它可在 schema mismatch 時回傳 Null；它不提供完整的逐列拒收紀錄，更不保證修好損毀 JSON。

明確 schema 也不等於「自動拒絕所有未知 key」。本篇選嚴格 allowlist：新欄位先隔離，確認語意後修改 intake 與 SCHEMA。這比靜默忽略欄位保守，但能避免折扣消失、營收看起來卻正常的問題。

十. 入口隔離：把來源位置留住，壞資料另行處理
#

把下面存成 intake.py。此為教學用的小型 admission gate：不強制轉型、不接受 bool 假裝整數；國家可缺省，商品可未知。

import hashlib
import json
from pathlib import Path


def validate(row):
    def require(ok, code):
        if not ok:
            raise ValueError(code)

    def text(value):
        return isinstance(value, str) and bool(value.strip())

    def integer(value):
        return type(value) is int and 0 <= value <= 1_000_000

    require(isinstance(row, dict), "object_required")
    require(set(row) <= {"order_id", "customer", "items", "tags"}, "extra_key")
    require(text(row.get("order_id")), "order_id_required")
    customer = row.get("customer")
    require(isinstance(customer, dict), "customer_object")
    require(set(customer) <= {"name", "country"}, "customer_extra")
    require(text(customer.get("name")), "customer_name")
    require(customer.get("country") is None or text(customer["country"]), "country")
    require("items" in row, "items_required")
    items = row["items"]
    require(items is None or isinstance(items, list), "items_list")
    if items is not None:
        for item in items:
            require(isinstance(item, dict), "item_object")
            require(set(item) == {"sku", "qty", "unit_cents"}, "item_fields")
            require(text(item["sku"]), "sku")
            require(integer(item["qty"]) and item["qty"] > 0, "qty")
            require(integer(item["unit_cents"]), "unit_cents")
    tags = row.get("tags", [])
    require(isinstance(tags, list) and all(text(t) for t in tags), "tags")
    return {**row, "customer": {"name": customer["name"],
            "country": customer.get("country")}, "tags": tags}


def admit(source: Path, accepted: Path, rejected: Path):
    counts = {"accepted": 0, "rejected": 0}
    with source.open(encoding="utf-8") as src, \
         accepted.open("w", encoding="utf-8") as good, \
         rejected.open("w", encoding="utf-8") as bad:
        for line_no, line in enumerate(src, 1):
            try:
                row = validate(json.loads(line))
                row["source_line"] = line_no
                good.write(json.dumps(row, ensure_ascii=False) + "\n")
                counts["accepted"] += 1
            except (ValueError, RecursionError) as error:
                code = "json_syntax" if isinstance(error, json.JSONDecodeError) else str(error)
                bad.write(json.dumps({"line": line_no, "code": code,
                    "sha256": hashlib.sha256(line.encode()).hexdigest()}) + "\n")
                counts["rejected"] += 1
    return counts

拒收檔只保存行號、錯誤代碼、雜湊，不複製整份 payload。原始來源要另行受控保存，才能用來源檔案與行號重新檢查。雜湊是核對用，不是匿名化或加密；不要拿它當隱私保證。跨檔合併時也要保存 batch/source ID，單一行號不是全域識別碼。

這段 gate 會逐行驗證，CPU 成本不是零；來源可信且契約穩定時可選批次驗證。輸入大小／深度限制、重複 JSON key 與 atomic publish 請另加正式邊界控制，不要把教學 gate 當成能直接接任意不可信 API 的完整防線。

十一. 重新掃描：主表與明細表分開
#

回到 demo.py，接上 intake，再只掃描合格檔：

from intake import admit
counts = admit(source, Path("accepted.ndjson"), Path("rejected.ndjson"))
assert counts == {"accepted": 3, "rejected": 0}
accepted_schema = {**SCHEMA, "source_line": pl.UInt64}
clean = pl.scan_ndjson("accepted.ndjson", schema=accepted_schema)
main_table = clean.select("order_id", "source_line", "customer", "tags", "items")
item_table = detail_query(clean)
assert main_table.collect().height == 3
assert item_table.collect().height == 2

沒有合格列時，先依 counts 分流；不要對空 NDJSON 檔猜 schema。正式輸出可保存兩張 Parquet：主表保留訂單與未知狀態，明細表只含真實商品。進入消費端前再驗證：訂單 ID 唯一、明細外鍵存在、金額界限符合契約。

十二. QA：不要只測「有沒有跑完」
#

存成 test_intake.py，用實際 drift 資料測 admission，而不是測 parser 能啟動：

import json
from intake import admit


def test_drift_quarantine(tmp_path):
    valid = {"order_id": "P001", "customer": {"name": "拍拍君"},
             "items": [{"sku": "book", "qty": 1, "unit_cents": 100}]}
    string_qty = json.loads(json.dumps(valid))
    string_qty["items"][0]["qty"] = "1"
    extra = {**valid, "discount": 50}
    source = tmp_path / "raw.ndjson"
    source.write_text("\n".join([json.dumps(valid), json.dumps(string_qty),
                                json.dumps(extra), "{broken"]) + "\n")
    good, bad = tmp_path / "good.ndjson", tmp_path / "bad.ndjson"
    assert admit(source, good, bad) == {"accepted": 1, "rejected": 3}
    assert json.loads(good.read_text())["source_line"] == 1
    failures = [json.loads(line) for line in bad.read_text().splitlines()]
    assert [f["code"] for f in failures] == ["qty", "extra_key", "json_syntax"]
    assert all("items" not in f for f in failures)