一. 前言：快，不該只靠換一台更大的機器 #
嗨，我是拍拍君。 你把一份幾 GB 的 Parquet 讀進來，接著篩選日期、挑五個欄位、做聚合， 最後只輸出幾百列。 但程式一開始就把整份資料塞進記憶體。 這不是資料太大，是執行順序太誠實了。 Polars 的 LazyFrame 會先記錄「想做什麼」，組成 logical plan， 等到 collect() 或 sink_*() 才最佳化並執行。 因此引擎有機會：
- 把 filter 推到資料來源旁邊
- 只掃描真正需要的欄位
- 把
head()變成更早發生的 slice - 合併可簡化的 expression
- 用 streaming engine 分批處理中間結果
這篇不是 Polars 入門。 如果你還不熟 select、filter、with_columns 與 Expression API， 可以先看Polars 基礎教學。 今天只追一件事：怎麼證明查詢真的照預期被最佳化。
二. 安裝與範例資料 #
本文以 Polars 1.44.2 為例：
uv add polars pyarrow
確認版本：
uv run python -c "import polars as pl; print(pl.__version__)"
先產生兩個小型 Parquet 分區，後面可以完整重跑：
from pathlib import Path
import polars as pl
out = Path("data/orders")
out.mkdir(parents=True, exist_ok=True)
for month in (8, 9):
rows = 50_000
frame = pl.DataFrame(
{
"order_id": range(month * 100_000, month * 100_000 + rows),
"customer_id": [i % 2_000 for i in range(rows)],
"region": ["north", "south", "east", "west"] * (rows // 4),
"status": ["paid", "paid", "cancelled", "refunded"]
* (rows // 4),
"amount": [(i % 500) + 0.5 for i in range(rows)],
"note": ["unused payload"] * rows,
"month": [month] * rows,
}
)
frame.write_parquet(out / f"orders-{month:02d}.parquet")
note 是故意放進去、後面又完全不用的欄位。 如果 projection pushdown 有生效，它不該進入主要運算路徑。
三. read_* 與 scan_*：最佳化從資料入口開始
#
先看 eager 寫法：
orders = pl.read_parquet("data/orders/*.parquet")
result = (
orders
.filter(pl.col("status") == "paid")
.select("region", "amount")
.group_by("region")
.agg(pl.col("amount").sum().alias("revenue"))
)
read_parquet() 會先 materialize 成 DataFrame。 後面的 filter 當然還能平行執行， 但檔案讀取已經發生，掃描階段錯過了最佳化資訊。 改成 scan_parquet()：
query = (
pl.scan_parquet("data/orders/*.parquet")
.filter(pl.col("status") == "paid")
.select("region", "amount")
.group_by("region")
.agg(pl.col("amount").sum().alias("revenue"))
.sort("revenue", descending=True)
)
現在 query 是 LazyFrame，還沒有讀完資料。 可以先取得 schema：
print(query.collect_schema())
最後才執行：
result = query.collect()
print(result)
實務原則很簡單：
| 需求 | 建議入口 |
|---|---|
| 小資料、互動探索 | read_csv() / read_parquet() |
| 可組合的正式管線 | scan_csv() / scan_parquet() |
| 已有 DataFrame，想延後執行 | df.lazy() |
df.lazy() 能最佳化後續 expression， 卻無法回到已完成的檔案掃描，所以不等於 scan_*()。
四. explain()：不要猜 optimizer 在想什麼
#
LazyFrame.explain() 會回傳查詢計畫文字。 先看未最佳化版本：
print(query.explain(optimized=False))
再看實際採用的最佳化計畫：
print(query.explain(optimized=True))
輸出格式會隨 Polars 版本變動， 不要把整段文字複製進 snapshot test。 閱讀時抓住四個問題就夠了：
SELECTION是否靠近 scan？- scan 是否只投影必要欄位？
SORT是否真的不可避免？- 是否出現重複、昂貴或不透明的節點？
也可以輸出 Graphviz 圖：
query.show_graph(optimized=True, output_path="query-plan.png")
這需要系統安裝 Graphviz；CI 不必為了圖片硬裝。
五. Predicate Pushdown：先丟掉不需要的列 #
Predicate pushdown 會把 filter 往資料來源推。
paid_orders = (
pl.scan_parquet("data/orders/*.parquet")
.filter(
(pl.col("status") == "paid")
& (pl.col("amount") >= 100)
)
.select("customer_id", "region", "amount")
)
對 Parquet 而言，引擎還能利用 row-group statistics， 跳過不可能符合條件的區塊。 但 pushdown 不是魔法。 如果先做不透明的 Python callback，optimizer 可能無法安全改寫：
def normalize(batch: pl.DataFrame) -> pl.DataFrame:
return batch.with_columns(
pl.col("status").str.to_lowercase()
)
opaque = (
pl.scan_parquet("data/orders/*.parquet")
.map_batches(normalize)
.filter(pl.col("status") == "paid")
)
能用 native expression，就別急著進 Python：
transparent = (
pl.scan_parquet("data/orders/*.parquet")
.with_columns(
pl.col("status").str.to_lowercase()
)
.filter(pl.col("status") == "paid")
)
native expression 不只通常更快， 也把型別與依賴關係留給 optimizer 判斷。
六. Projection Pushdown：不要把沒用的欄位搬進來 #
Projection pushdown 會從最終輸出反推真正需要的欄位。
summary = (
pl.scan_parquet("data/orders/*.parquet")
.filter(pl.col("status") == "paid")
.group_by("region")
.agg(
pl.col("amount").sum().alias("revenue"),
pl.col("order_id").n_unique().alias("orders"),
)
)
這條管線只需要：
- filter 用的
status - group key
region - 聚合用的
amount與order_id
customer_id、note、month 不必讀進主要計畫。 別為了「寫得保險」先 select(pl.all())， 也不要太早把整列包成 struct 再交給自訂函式。 最終需要什麼，就用 expression 明確說出來。
七. Streaming engine：分批執行，不等於逐列 Python loop #
Polars 1.44 使用 engine="streaming" 明確選擇 streaming engine：
result = query.collect(engine="streaming")
舊文章常見：
# 舊 API 寫法，不建議新程式照抄
result = query.collect(streaming=True)
Streaming 的重點是讓引擎以 batch 處理可串流節點， 降低大型中間結果的記憶體壓力。 它不是保證「任何檔案都永遠不會 OOM」。 全域排序、高基數 group-by、某些 join， 仍可能需要大量狀態或發生 fallback。 開啟 verbose 訊息可以協助診斷：
POLARS_VERBOSE=1 uv run python pipeline.py
不要把「有加 streaming」當完成條件。 要同時量測：
- peak memory
- wall-clock time
- 輸出列數與 checksum
- 是否發生 engine fallback
八. sink_*()：結果很大時，不要先 collect 再寫檔
#
這段看似合理：
result = query.collect(engine="streaming")
result.write_parquet("output/revenue.parquet")
但 result 仍需成為記憶體中的完整 DataFrame。 如果目的地本來就是檔案，使用 native sink：
query.sink_parquet(
"output/revenue.parquet",
compression="zstd",
mkdir=True,
engine="streaming",
)
也有 CSV 與 NDJSON sink：
query.sink_csv(
"output/revenue.csv",
mkdir=True,
engine="streaming",
)
原則是：
最終需要 DataFrame 才
collect()；最終需要檔案就sink_*()。
collect_batches() 可以把結果交給自訂 batch consumer， 但官方文件提醒它比 native sink 慢。 只有目的地沒有對應 sink 時才考慮：
for batch in query.collect_batches(chunk_size=10_000):
send_to_custom_service(batch)
九. Benchmark：計畫合理，還要量實際成本 #
Polars 1.44.2 已將舊的 LazyFrame.profile() 標成 deprecated， 因為 concurrent streaming engine 的節點時間容易產生誤導。 新管線先量端到端時間：
from time import perf_counter
started = perf_counter()
result = query.collect(engine="streaming")
elapsed = perf_counter() - started
print(f"rows={result.height:,} elapsed={elapsed:.3f}s")
請用接近正式環境的資料測，別拿 100 列 sample 就宣告最佳化成功。 測量時固定：
- Polars 版本
- input files 與 schema
- cold cache / warm cache 條件
- engine 設定
- 輸出排序要求
不要隨便插入 cache()。 快取會改變計畫與記憶體生命週期， 應該先用可重複 benchmark 證明重算真的是瓶頸。
十. 一條可維護的 LazyFrame 管線 #
把查詢建立與執行分開，測試會容易很多：
from pathlib import Path
import polars as pl
def build_revenue_query(source: str | Path) -> pl.LazyFrame:
return (
pl.scan_parquet(source)
.filter(
(pl.col("status") == "paid")
& (pl.col("amount") >= 100)
)
.select("region", "customer_id", "amount")
.group_by("region")
.agg(
pl.col("amount").sum().round(2).alias("revenue"),
pl.col("customer_id").n_unique().alias("customers"),
)
.sort("revenue", descending=True)
)
def main() -> None:
query = build_revenue_query("data/orders/*.parquet")
print(query.explain())
query.sink_parquet(
"output/revenue.parquet",
compression="zstd",
mkdir=True,
engine="streaming",
)
if __name__ == "__main__":
main()
函式回傳 LazyFrame，呼叫端可以：
- 在開發時
explain() - 在 benchmark 時包住 materialization 計時
- 在 notebook 裡
collect() - 在 production 用
sink_parquet()
執行策略不必混進每個 transformation。
十一. 測試：驗證結果，也監看計畫形狀 #
先測語意，不要只測速度：
from pathlib import Path
import polars as pl
from polars.testing import assert_frame_equal
from pipeline import build_revenue_query
def test_revenue_query(tmp_path: Path) -> None:
source = tmp_path / "orders.parquet"
pl.DataFrame(
{
"region": ["north", "north", "south"],
"customer_id": [1, 2, 3],
"status": ["paid", "cancelled", "paid"],
"amount": [150.0, 999.0, 80.0],
}
).write_parquet(source)
actual = build_revenue_query(source).collect()
expected = pl.DataFrame(
{
"region": ["north"],
"revenue": [150.0],
"customers": [1],
},
schema={
"region": pl.String,
"revenue": pl.Float64,
"customers": pl.UInt32,
},
)
assert_frame_equal(actual, expected)
計畫文字不承諾穩定格式，適合做低脆弱度 smoke check， 而不是整段 golden file；版本升級時要先讀新 plan，別機械式更新字串。
十二. 常見反最佳化寫法 #
12.1 太早 collect()
#
df = pl.scan_parquet(path).filter(predicate).collect()
result = df.lazy().group_by("region").agg(pl.col("amount").sum())
這會把一條可整體最佳化的管線切成兩段。
12.2 用 Python UDF 取代 native expression #
Python function 往往較慢，也可能成為 optimizer barrier。 先找 pl.col().str、.dt、.list、.struct 與 built-in expression。
12.3 為了穩定輸出，到處 sort #
Sort 是昂貴的全域操作。 只在語意或最終輸出真的需要順序時排序。
12.4 把 streaming 當萬靈丹 #
Streaming 降低某些計畫的記憶體壓力， 不會消除所有 stateful operation 的成本。
12.5 只看執行時間，不驗證答案 #
最快的錯誤答案還是錯誤答案。 效能測試必須搭配 row count、schema、關鍵總和與 sample checksum。
十三. 上線前檢查清單 #
- 檔案入口使用
scan_*()，而不是先read_*() -
explain()中 filter 靠近 scan - scan 只讀需要的欄位
- 沒有可避免的 Python UDF
- 全域 sort、join 與 group-by 都有明確理由
- 大型輸出優先使用
sink_*() - explicit streaming 與預設 engine 都做過 benchmark
-
POLARS_VERBOSE=1沒有被忽略的 fallback - 測試同時驗證 schema、列數與關鍵聚合
- 升級 Polars 後重新檢查 plan 與 benchmark
結語：最佳化要能被觀察，也要能被重跑 #
LazyFrame 真正有價值的地方，不是把 .lazy() 塞進 method chain。 它讓資料入口、expression、optimizer 與執行引擎共享完整計畫。 今天最值得帶走的工作流是：
- 用
scan_*()保留資料來源資訊 - 用 native expression 描述 transformation
- 用
explain()檢查 pushdown 與昂貴節點 - 用可重複 benchmark 量實際時間與記憶體
- 用
collect(engine="streaming")或sink_*()選擇執行方式 - 用結果測試與 plan smoke check 防止退化
別再用「Polars 應該很快吧」當效能保證。 讓計畫說話，讓 benchmark 作證。拍拍君下次見。