Lesson 05 · 工程

写好一个数据集：分区、大小、演进

一个文件变成三千个文件，体积翻倍、查询慢 50 倍。分区粒度是最容易犯的错。

分区：把过滤条件变成目录名

所谓 Hive 风格分区，就是把某一列的值编进目录名，让引擎连文件都不用打开就能排除：

warehouse/iot/
├── day=2026-01-01/
│   └── data_0.parquet      ← 这一天的全部数据
├── day=2026-01-02/
│   └── data_0.parquet
└── day=2026-01-03/
    └── data_0.parquet

查询 WHERE day='2026-01-02'
  → 引擎看目录名就知道只需打开一个文件，其余连 footer 都不读

这是比 row group 统计更粗、也更便宜的一层过滤：第 4 课的四层跳过发生在文件内部，分区裁剪发生在文件之外。两者叠加，不冲突。

COPY (SELECT *, strftime(ts, '%Y-%m-%d') AS day
      FROM 'data/iot_sorted.parquet'
      ORDER BY device_id, ts)                -- 分区内仍要排序！
  TO 'warehouse/iot'
  (FORMAT parquet, PARTITION_BY (day), COMPRESSION zstd, ROW_GROUP_SIZE 200000);

-- ** 匹配任意层目录；hive_partitioning 让 day 变成一个可查询的列
SELECT day, count(*), round(avg(value), 3)
FROM read_parquet('warehouse/iot/**/*.parquet', hive_partitioning = true)
WHERE day BETWEEN '2026-01-05' AND '2026-01-08'
GROUP BY 1 ORDER BY 1;

实测：分区切太细会发生什么

同一份 4,320,000 行数据，三种切法：

第三行就是小文件灾难的教科书样本。文件数涨 200 倍，代价是：

• 体积翻倍（14.16 → 31.42 MB）。每个文件都要带一份 footer、一份 schema、每列一套字典和统计。10 KB 的文件里，元数据占了很大比例，而且字典被切碎后压缩效果直线下降。
• 查询慢 13–50 倍。哪怕是"只查一台设备"这种理论上最该受益于细分区的查询，也从 3.1 ms 变成 121.6 ms——因为引擎要 open / 读 footer / close 三千多次。在 S3 上这个代价还要再乘一个数量级（每个文件至少两次网络往返）。

分区键的作用是把数据切成"查询通常一次要一整块"的粒度，不是把数据切碎。第二个分区键几乎总是错的。

分区粒度怎么定

一条经验规则，够用：让每个分区里的文件落在 100 MB – 1 GB 量级。反推分区粒度：

本课样本每天只有 ~900 KB，按天分区其实已经偏细了——它在表里看起来赢，是因为数据集太小、全在本地磁盘上。真实的 Pivot 场景要按实际日增量反推。

单文件与 row group 该多大

官方 Configurations 给的建议是 row group 512 MB – 1 GB、data page 8 KB。但要看清它的前提：

"Since an entire row group might need to be read, we want it to completely fit on one HDFS block."
——因为可能要读整个 row group，我们希望它正好装进一个 HDFS block。

这是 HDFS 时代的取值。今天绝大多数人把 Parquet 放在 S3 / OSS 上，没有 HDFS block 这回事，而单机引擎（DuckDB / Polars）的并行单位就是 row group——row group 太大反而没法并行，也没法跳过（第 4 课那个 5 个 rg 只能跳掉一点点的例子）。

小文件治理：compaction

即使分区定对了，流式写入也会产生小文件——每小时一批就是每天 24 个文件。标准做法是定期合并（compaction）：把一个分区里的碎片读进来、排序、重新写成一两个大文件。

-- 把 day=2026-01-08 下的所有碎片合并成一个排好序的文件
COPY (SELECT * EXCLUDE (day)
      FROM read_parquet('warehouse/iot/day=2026-01-08/*.parquet')
      ORDER BY device_id, ts)
  TO 'warehouse/iot/day=2026-01-08/compacted.parquet'
  (FORMAT parquet, COMPRESSION zstd, ROW_GROUP_SIZE 200000);
-- 确认无误后再删除原碎片（顺序不能反！）

Schema 演进：加了一列怎么办

设备固件升级，上报里多了一个 firmware 字段。老文件没有这列，新文件有。混着读会怎样？

-- 默认：按位置对齐列，schema 不一致时可能报错或错位
SELECT * FROM read_parquet('warehouse/iot/**/*.parquet');

-- 正确姿势：按列名合并，缺的列自动补 NULL
SELECT * FROM read_parquet('warehouse/iot/**/*.parquet', union_by_name = true);

实测：老文件 1000 行（7 列）+ 新文件 1000 行（8 列），union_by_name=true 读出 2000 行，其中 firmware 非空 1000 行——老数据那部分自动补 NULL。

裸 Parquet 目录能安全承受的演进只有这几种：

真正的 schema 演进管理（列重命名、类型变更、字段 ID 追踪）是表格式的职责，不是 Parquet 的。这也是判断"该不该上 Iceberg"的第二个信号。

import duckdb, os, glob, shutil, time
con = duckdb.connect()
con.sql("CREATE VIEW src AS SELECT *, strftime(ts,'%Y-%m-%d') AS day "
        "FROM 'data/iot_sorted.parquet'")

for d in ["data/p_day", "data/p_day_dev"]: shutil.rmtree(d, ignore_errors=True)
con.sql("COPY (SELECT * FROM src ORDER BY device_id, ts) TO 'data/p_day' "
        "(FORMAT parquet, PARTITION_BY (day), COMPRESSION zstd, ROW_GROUP_SIZE 200000)")
con.sql("COPY (SELECT * FROM src ORDER BY device_id, ts) TO 'data/p_day_dev' "
        "(FORMAT parquet, PARTITION_BY (day, device_id), COMPRESSION zstd, ROW_GROUP_SIZE 200000)")

for d, g in [("p_day", "data/p_day/*/*.parquet"),
             ("p_day_dev", "data/p_day_dev/*/*/*.parquet")]:
    fs = glob.glob(g); tot = sum(os.path.getsize(f) for f in fs)
    s = time.perf_counter()
    con.sql(f"SELECT avg(value) FROM read_parquet('{g}', hive_partitioning=true) "
            f"WHERE device_id='AHU-05-013'").fetchall()
    print(f"{d:10s} 文件 {len(fs):5d}  总计 {tot/1024/1024:6.2f} MB  "
          f"单设备查询 {(time.perf_counter()-s)*1000:7.1f} ms")

随时问我：把你要归档的表的日增量大小和最常见的三个查询告诉我，我直接帮你算分区粒度和 row group 大小，并写出对应的 COPY 语句。这是本课程最实用的一次对话。