9.1 从单机越界到分区计算：先缩小工作集，再增加机器

十座站点产生了五年的设备日志，压缩文件共 500 GB。pd.read_csv() 很快耗尽内存，但这不等于必须立刻搭集群。所需结果只是按月、设备型号和故障频率聚合，实际工作集可以远小于原始数据。

数据规模没有一条固定的“16GB 用 pandas、100GB 用 Spark”分界。工具选择取决于工作集、算子、存储布局、并发、时限和团队运维能力。

本课目标

区分数据集大小、工作集与结果大小；
判断哪些计算可以安全分块合并；
利用列式格式、列裁剪和过滤下推减少 I/O；
用证据决定继续单机、使用并行框架还是上集群。

1. 先定位瓶颈

一次慢查询可能受限于：

内存：中间表、哈希表或排序无法驻留；
I/O：读取大量无关列和行；
CPU：解析文本、压缩、Python 函数或复杂表达式；
网络：远程读取或节点间 shuffle；
调度：大量微小文件和任务；
倾斜：少数键让一个分区成为长尾。

记录峰值内存、扫描字节、CPU 利用率、I/O 吞吐、任务分布和结果行数。只看原文件大小会误判：20GB join 可能产生数 TB 中间结果，500GB 日志也可能经分区裁剪后只读几 GB。

2. 工作集决定内存压力

工作集是某个执行阶段需要同时保留的数据。读取可以流式进行，不代表所有算子都能流式：

行级过滤、投影、逐行转换通常可分区执行；
sum、count、min、max 可先算局部状态再合并；
精确全局排序、某些窗口和高基数去重可能需要大规模重排；
中位数、分位数和任意 Python UDF 是否可分块，取决于算法与实现。

分块计算必须有正确的合并规则。例如总体均值不能直接平均各块均值；应合并 sum 和 count：

python

from io import StringIO
import pandas as pd

csv_data = "mission_type,resources\nscan,10\nscan,20\nscan,90\nrepair,\n,5\n"

def aggregate_chunks(source, chunk_size=2):
    merged = pd.DataFrame(columns=["sum", "count"], dtype=float)
    for chunk in pd.read_csv(source, dtype={"mission_type": "string", "resources": "float64"},
                             usecols=["mission_type", "resources"], chunksize=chunk_size):
        part = chunk.groupby("mission_type", dropna=False)["resources"].agg(["sum", "count"])
        merged = pd.concat([merged, part]).groupby(level=0, dropna=False).sum()
    merged["mean"] = merged["sum"] / merged["count"].where(merged["count"] > 0)
    return merged

merged = aggregate_chunks(StringIO(csv_data))
print(merged.loc["scan", ["sum", "count", "mean"]].tolist())
# [120.0, 3.0, 40.0]
assert pd.isna(merged.loc["repair", "mean"])

示例只依赖 pandas，缺失分组键保留为一组，资源缺失不计入均值分母。scan 两块均值为 15 和 90，直接平均得到错误的 52.5；合并总和 120 与计数 3 才得到 40。只保留累计聚合状态可避免囤积所有块，但内存仍随分组键数量增长，高基数聚合也可能越界。

3. 先改变存储和查询方式

CSV 适合交换，却缺少可靠类型、列级读取和丰富统计信息。对反复分析的数据，可转换为带显式 schema 的列式格式。

下面在临时目录创建同一份 Parquet 数据，需另装 pyarrow、duckdb 和 dask[dataframe]，本页 Python 块按顺序运行：

python

from pathlib import Path
from tempfile import TemporaryDirectory

work = TemporaryDirectory(prefix="atlas-partitions-")
data_path = Path(work.name) / "missions"
fixture = pd.DataFrame({
    "event_date": pd.to_datetime(["2025-12-31", "2026-01-01", "2026-01-02"]),
    "mission_type": ["scan", "scan", "repair"],
    "resources": [100.0, 10.0, 30.0],
    "year": [2025, 2026, 2026], "month": [12, 1, 1],
})
fixture.to_parquet(data_path, engine="pyarrow", partition_cols=["year", "month"], index=False)
# Keep work alive until the SQL and Dask examples finish.

Parquet 等列式格式允许执行引擎只读需要的列；若文件/row group 元数据支持，还能把过滤推到扫描端并跳过不可能匹配的数据。实际收益取决于压缩、选择率、布局、存储和引擎，不能承诺固定“快 5–10 倍”。

sql

SELECT mission_type, avg(resources) AS mean_resources
FROM read_parquet('PATH/year=*/month=*/*.parquet', hive_partitioning = true)
WHERE year = 2026 AND event_date >= TIMESTAMP '2026-01-01'
GROUP BY mission_type;

在本地分析引擎中直接查询文件，常比先把全部数据载入 DataFrame 更省内存。

SQL 是 DuckDB 方言，PATH 换成上面 data_path 的实际路径；也可直接运行：

python

import duckdb

sql_result = duckdb.sql("""
    SELECT mission_type, avg(resources) AS mean_resources
    FROM read_parquet(?, hive_partitioning = true)
    WHERE year = 2026 AND event_date >= TIMESTAMP '2026-01-01'
    GROUP BY mission_type ORDER BY mission_type
""", params=[str(data_path / "year=*" / "month=*" / "*.parquet")]).fetchall()
print(sql_result)
# [('repair', 30.0), ('scan', 10.0)]

year = 2026 是目录分区谓词，event_date 则针对文件内数据。仅写日期过滤并不意味着引擎一定推导出目录年份条件。用执行计划和扫描统计分别确认分区裁剪与 row-group 跳过。

4. 物理分区服务于常用过滤

按日期、区域等字段组织目录，可以让引擎跳过无关分区。但分区列不是越多越好：

高基数键会制造海量目录和小文件；
查询不使用分区谓词时，仍需扫描大量文件；
小文件增加列表、打开和调度开销；
写入过度并行会把每个分区切成碎片。

选择少量、稳定、常用于粗粒度过滤的字段。文件内部的排序和 row-group 统计可进一步帮助跳过数据。定期 compaction 需要保持原子发布和读写并发安全。

5. 延迟执行不是自动更快

Dask DataFrame 等框架把操作构造成任务图，用到结果时才执行：

python

import dask.dataframe as dd

ddf = dd.read_parquet(
    data_path,
    columns=["event_date", "mission_type", "resources"],
    filters=[[("event_date", ">=", pd.Timestamp("2026-01-01"))]],
)
result = ddf.groupby("mission_type")["resources"].agg(["sum", "count"])
final = result.compute()
print(final.sort_index().to_dict("index"))
# {'repair': {'sum': 30.0, 'count': 1}, 'scan': {'sum': 10.0, 'count': 1}}
work.cleanup()

延迟执行让优化器有机会做列裁剪、过滤下推和任务融合，也避免过早物化中间结果。但：

过小分区会产生巨大任务图；
过大分区会让多个并行任务共同耗尽 worker 内存；
循环中反复 compute() 会拆散优化机会；
pandas API 相似不代表执行成本与支持范围相同。

先用 dashboard 或执行报告观察任务、内存、spill 和传输，而不是猜一个分区数。

本例只在本地执行小数据并校对结果，不构成吞吐基准。compute() 会把最终结果收回调用进程；即使输入分布式，巨大的结果仍可能让客户端 OOM。Parquet 元数据读取和分区选择的细节见 Dask 官方说明。

6. 什么时候引入分布式

下面是决策维度，不是容量阈值：

问题	单机/嵌入式引擎更合适	分布式框架更合适
工作集	可流式或可在磁盘溢写内完成	单机时限或磁盘带宽不可接受
计算	扫描、过滤、常规聚合	多次大 shuffle、并行 ETL、迭代作业
并发	一名分析者、少量任务	多团队共享、需要队列和资源隔离
数据位置	本地或单个对象存储区域	数据已在集群/分布式存储中
SLA	分钟到小时可接受	需要更高吞吐或稳定截止时间
运维	希望少组件	已有调度、监控、安全与值班能力

扩大机器、减少扫描、预聚合和改算法都可能比增加节点便宜。分布式解决的是可并行工作的吞吐与容量，不会消除网络和协调成本。

7. 迁移前做一组基准

选择有代表性的查询，而不是只测试 count(*)：

text

窄表过滤与聚合
高基数 group by
大表与小维表 join
两张大表 join
窗口或排序
最终写出与小文件数量

对每组记录输入规模、扫描字节、峰值内存、执行时间、成本和结果校验值。冷缓存与热缓存分开测，避免把缓存收益误认为引擎收益。

常见误区

数据大于内存就必须上集群：可流式、磁盘溢写和列裁剪可能足够。
换 Parquet 必然有固定倍数提升：收益取决于查询与布局。
分块后平均各块均值：块大小不同会产生错误结果。
API 相似即可机械迁移：分区、索引、shuffle 和 UDF 成本不同。

练习

把一个 OOM 查询拆成扫描、过滤、聚合和结果四个工作集。
为分块方差计算设计可合并状态，而不是平均块方差。
比较 CSV 全读与 Parquet 列裁剪的扫描字节和峰值内存。
用同一业务查询比较单机 SQL 引擎与分布式框架，并写出迁移结论。

小结

“数据有多大”只是起点。决策依据是每个阶段的工作集、数据移动量和服务目标。先减少不必要的读取与中间结果，再决定是否值得承担分布式复杂度。

下一课拆开分布式批处理的成本结构：分区带来并行，shuffle、倾斜和重试则决定作业能否稳定完成。

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算

9.1 从单机越界到分区计算：先缩小工作集，再增加机器

本课目标

1. 先定位瓶颈

2. 工作集决定内存压力

3. 先改变存储和查询方式

4. 物理分区服务于常用过滤

5. 延迟执行不是自动更快

6. 什么时候引入分布式

7. 迁移前做一组基准

常见误区

练习

小结