14  Dask并行计算

14.1 先修知识与学习目标

本章假定读者掌握 Pandas 分组聚合、索引和 Parquet 读取，并理解延迟执行。Dask DataFrame 把许多 Pandas DataFrame 组织成分区，通过任务图调度；它并非把任意 Pandas 代码自动改写成高效分布式程序，也不保证数据超出内存时所有操作都安全。

完成本章后，读者应能：

  • 读取和解释 Dask 的任务图，区分逻辑依赖与调度执行；
  • 说明分区、索引与 divisions 的关系，并判断顺序相关操作是否有依据；
  • 识别 set_index、全局排序、连接和高基数分组引起的 shuffle；
  • 根据 Python GIL、数据局部性与集群规模选择线程、进程或 distributed 调度器；
  • 为 map_partitions() 提供正确 meta，并控制 .compute()、.persist() 与输出物化的内存边界。

本章用下表把每项正式目标落实到可观察的课堂活动和核心答案。

表 14.1: 第十四章目标—活动—核心题映射
正式目标 正文活动 核心题与答案证据
任务图与延迟执行 小节 14.2 练习 14.1 的关键路径计算与终端动作解释
分区、索引与 divisions 小节 14.4 练习 14.2 的日期边界断言
Shuffle 识别 小节 14.6 练习 14.4 的组合键路由解释与 Pandas 核验
调度器选择 小节 14.7 练习 14.5 的三个工作负载决策
meta 与物化内存 小节 14.8、小节 14.9 练习 14.3 的 schema 和四行结果物化审计

表 表 14.1 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。

14.2 任务图：从表达式到执行

Dask 把计算表示为有向无环图 \(G=(V,E)\)。节点 \(v\in V\) 是读取分区、筛选、局部聚合或合并等任务；边 \((u,v)\in E\) 表示任务 \(v\) 依赖 \(u\) 的输出。调度器只能在依赖满足后运行节点。

若任务 \(v\) 的耗时为 \(w_v\)，总工作量 \(W=\sum_v w_v\)，关键路径长度为 \(L\)，使用 \(p\) 个 worker 的完成时间满足：

\[ T_p \geq \max\left(\frac{W}{p},L\right). \tag{14.1}\]

式 14.1 说明：增加核心数不能缩短必须串行的关键路径，过细任务还会增加调度开销。

分区聚合的任务图 读取层 局部计算层 合并层 分区 0分区 1分区 2 筛选 + 局部和筛选 + 局部和筛选 + 局部和 组合 A组合 B最终结果
图 14.1: Dask 对三个输入分区分别筛选并计算部分和，再以树形归约合并；箭头表示依赖，而不是固定的线程执行顺序。

图 14.1 展示了可并行的局部阶段和较短的归约阶段。任务图是惰性描述；只有 .compute()、.persist()、写出或部分预览等终端动作才触发执行。

14.3 真实 Parquet 数据与初始分区

本章使用本地 2023 年全市场行情 Parquet 分片。一个 Dask 分区通常对应一个或多个文件片段，但分区并不等于证券、交易日或 SQL 分组。

import platform  # 识别运行平台以选择规范数据路径
from pathlib import Path  # 构造本地数据集路径
import dask  # 配置与检查Dask任务调度
import dask.dataframe as dd  # 构建分区化DataFrame计算
import pandas as pd  # 定义分区函数输出的meta

DATA_ROOT = 'C:/qiufei/data' if platform.system() == 'Windows' else '/home/ubuntu/r2_data_mount/data'  # 选择跨平台数据根目录
MARKET_DATASET = Path(DATA_ROOT) / 'stock' / 'full_market_with_return_2023.parquet'  # 指向真实行情Parquet目录
YRD_CODES = ['600104.XSHG', '600276.XSHG', '002415.XSHE', '002142.XSHE']  # 固定长三角公司样本
MARKET_FILES = [MARKET_DATASET / file_name for file_name in ['part.0.parquet', 'part.1.parquet', 'part.8.parquet', 'part.10.parquet']]  # 使用schema审计确认包含目标证券的四个物理分片
MARKET_COLUMNS = ['order_book_id', 'trade_date', 'high', 'low', 'adj_close', 'volume', 'total_turnover', 'daily_return']  # 仅读取教学所需列
dask_market = dd.read_parquet(MARKET_FILES, columns=MARKET_COLUMNS, filters=[[('order_book_id', 'in', YRD_CODES), ('trade_date', '<', pd.Timestamp('2024-01-01'))]])  # 对目标物理分片下推证券与年度筛选
dask_yrd = dask_market  # 为后续任务图保留语义清晰的长三角样本名称
print(f'Dask版本为 {dask.__version__}')  # 记录实际执行版本
print(f'初始分区数为 {dask_yrd.npartitions}')  # 报告文件布局形成的分区数
print(dask_yrd.dtypes)  # 核验真实数据schema
Dask版本为 2025.10.0
初始分区数为 4
order_book_id     string[pyarrow]
trade_date         datetime64[ns]
high                      float64
low                       float64
adj_close                 float64
volume                    float64
total_turnover            float64
daily_return              float64
dtype: object

筛选后分区数通常不变：有些分区可能只剩少量行甚至为空。要判断分区是否失衡，应执行 map_partitions(len).compute()，但这会扫描数据。

14.4 分区与 divisions

divisions=(d_0,d_1,\ldots,d_p) 描述 \(p\) 个按索引有序且相邻的分区边界：第 \(i\) 个分区索引大致落在 \([d_i,d_{i+1})\)，最后一个分区包含右端点。若边界未知，Dask 用一串 None 表示。已知 divisions 能支持按索引定位、切片和某些对齐；它不表示每个分区行数相等。

print(f'是否已知divisions：{dask_yrd.known_divisions}')  # 判断索引边界是否可用于有序定位
print(dask_yrd.divisions)  # 展示每个分区的边界元数据
partition_sizes = dask_yrd.map_partitions(len, meta=('row_count', 'int64')).compute()  # 扫描各分区以度量过滤后的实际行数
print(partition_sizes.describe())  # 报告分区大小离散程度
是否已知divisions：False
(None, None, None, None, None)
count      2.0
mean     482.0
std        0.0
min      482.0
25%      482.0
50%      482.0
75%      482.0
max      482.0
Name: row_count, dtype: float64

若要按交易日切片，可用 set_index('trade_date') 建立有序索引。对无序输入，这通常需要采样边界并执行全局 shuffle；不能把它当作免费元数据操作。

date_indexed = dask_yrd.set_index('trade_date', npartitions=4, shuffle_method='tasks')  # 通过任务式shuffle重排为日期索引分区
print(f'设置索引后是否已知divisions：{date_indexed.known_divisions}')  # 检查采样与重排得到的边界状态
print(date_indexed.divisions)  # 输出日期分区边界
december_preview = date_indexed.loc['2023-12-01':'2023-12-31'].head(8)  # 利用已知日期边界只预览12月少量记录
december_preview  # 展示按索引切片结果
设置索引后是否已知divisions：True
(Timestamp('2023-01-04 00:00:00'), Timestamp('2023-03-22 00:00:00'), Timestamp('2023-06-20 00:00:00'), Timestamp('2023-09-19 00:00:00'), Timestamp('2023-12-29 00:00:00'))
order_book_id high low adj_close volume total_turnover daily_return
trade_date
2023-12-01 002142.XSHE 52.5686 51.5815 51.9258 2.433296e+07 8.586119e+08 -0.013090
2023-12-01 002415.XSHE 793.9646 771.5994 793.9646 1.125238e+06 7.091568e+08 0.020408
2023-12-01 600104.XSHG 159.5255 156.9542 159.2041 6.086210e+06 3.955746e+08 -0.001344
2023-12-01 600276.XSHG 3113.2602 2964.4861 3012.9994 6.936388e+05 1.904556e+09 -0.029987
2023-12-04 002142.XSHE 51.9487 49.7680 50.3878 3.556135e+07 1.219197e+09 -0.029619
2023-12-04 002415.XSHE 804.9235 780.0982 780.5455 1.104499e+06 7.015617e+08 -0.016901
2023-12-04 600104.XSHG 160.2755 157.3828 157.5971 4.283501e+06 2.780083e+08 -0.010094
2023-12-04 600276.XSHG 3016.8804 2943.1402 2974.8356 4.704981e+05 1.277259e+09 -0.012666

同一天有多只证券，因此日期索引不唯一；这不妨碍按日期切片，但任何需要证券内部顺序的计算仍要按证券和日期明确组织。

14.5 局部任务、聚合与图规模

可分解聚合通常包含 chunk、combine、aggregate 三层。求和维护每组部分和，工作量约为 \(O(n)\)；若输出分为 \(s\) 个分区，按非索引键分组可能引入 hash shuffle 和约 \(O(n)\) 的数据传输。

turnover_by_stock_lazy = dask_yrd.groupby('order_book_id')['total_turnover'].sum(split_out=2)  # 构建带两个输出分区的哈希分组聚合
task_count = len(turnover_by_stock_lazy.__dask_graph__())  # 统计当前优化表达式对应的任务节点数
print(f'聚合前仍是延迟对象，任务数为 {task_count}')  # 说明尚未物化业务结果
聚合前仍是延迟对象，任务数为 36
with dask.config.set(scheduler='threads'):  # 对释放GIL的Parquet解码与数值聚合采用线程调度器
    turnover_by_stock = turnover_by_stock_lazy.compute().sort_values(ascending=False)  # 只把四行聚合结果带回驱动进程
turnover_by_stock.to_frame('annual_turnover')  # 展示真实公司年度成交额
表 14.2: Dask线程调度器计算的长三角公司年度成交额
annual_turnover
order_book_id
600276.XSHG 3.781830e+11
002415.XSHE 3.286586e+11
002142.XSHE 1.993612e+11
600104.XSHG 6.962311e+10

这里 .compute() 安全，是因为结果仅四行；若对全市场明细调用 .compute()，所有分区会汇集到驱动进程并转成单个 Pandas 对象，可能超过内存。

14.5.1 结果解释

表 14.2 的排序描述四家公司在本地样本期的成交活跃度。结果应与第 11–13 章的同口径统计在数据覆盖一致时相符；它验证的是分区聚合语义，而不是 Dask 必然更快。对仅约千行的筛选结果，调度开销可能高于单机 Pandas，Dask 的教学价值在于展示任务规模增大时仍可保持同一分区化执行模型。

14.6 Shuffle：最昂贵的语义转换之一

当相同键的行分散在多个输入分区，而下游算子要求同键共置时，需要 shuffle。典型触发器包括：

  • 对非索引键执行高基数分组并产生多个输出分区；
  • set_index() 和全局 sort_values()；
  • 连接键与已知 divisions 不对齐的 merge；
  • 去重和某些累计、窗口操作。

设输入字节量为 \(B\)，shuffle 至少要检查并路由这些数据，通信或本地磁盘 I/O 量通常为 \(O(B)\)。实际时间还受分区倾斜、网络、序列化与打开文件数影响。先筛选列和行、选择稳定分区键、避免过多微小分区，往往比盲目增加 worker 更有效。

错误用法对照：先制造宽表，再全局排序

# 该片段仅展示反模式，不在教材中执行
wide_market = dask_market.merge(another_large_table, on='order_book_id')  # 在筛选前连接会扩大shuffle字节量
sorted_market = wide_market.sort_values('total_turnover')  # 全局排序再次触发昂贵的数据重排
all_rows = sorted_market.compute()  # 将全量宽表汇集到驱动进程可能耗尽内存

正确流程：先下推筛选与投影，再聚合或分区写出

compact_market = dask_market[['order_book_id', 'trade_date', 'total_turnover']]  # 先投影为窄表以减少传输字节
compact_yrd = compact_market[compact_market['order_book_id'].isin(YRD_CODES)]  # 在重排前筛掉无关证券
monthly_turnover_lazy = compact_yrd.assign(trade_month=compact_yrd['trade_date'].dt.to_period('M').astype(str)).groupby(['order_book_id', 'trade_month'])['total_turnover'].sum(split_out=2)  # 仅对必要数据执行月度哈希聚合
monthly_turnover = monthly_turnover_lazy.compute().sort_index()  # 物化约四十八行结果而非日度明细
monthly_turnover.head(12)  # 展示一个证券年度附近的月度结果
order_book_id  trade_month
002142.XSHE    2023-01        1.400670e+10
               2023-02        1.818892e+10
               2023-03        2.090744e+10
               2023-04        2.306258e+10
               2023-05        1.637285e+10
               2023-06        1.142916e+10
               2023-07        1.859636e+10
               2023-08        2.031329e+10
               2023-09        1.462708e+10
               2023-10        1.312367e+10
               2023-11        1.243779e+10
               2023-12        1.629536e+10
Name: total_turnover, dtype: float64

14.7 调度器选择

表 14.3: Dask 调度器的适用边界
调度器 地址空间 适合任务 主要代价
synchronous 单线程 调试、复现异常 无并行
threads 单进程共享内存 NumPy、Pandas、Arrow 等释放 GIL 的计算与 I/O Python 字节码受 GIL 约束
processes 多进程隔离内存 不释放 GIL 的 CPU 密集 Python 函数 序列化与进程内存复制
distributed 单机或集群 worker 监控、容错、跨机数据 网络、部署与数据局部性成本

表 表 14.3 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。

Dask DataFrame 的列式数值算子常在原生代码中释放 GIL，因此线程是合理起点；大量纯 Python UDF 才可能受益于进程，但序列化大分区可能抵消收益。Distributed 调度器不等同于自动扩展：数据位置、worker 内存阈值、spill 配置和网络拓扑都必须设计。

14.8 meta：不执行数据也要知道输出结构

Dask 在建图时需要列名、顺序与 dtype。若不给 meta，它可能用伪造的小样本调用函数推断结构；具有数据依赖分支或副作用的函数会因此失败或产生错误 schema。应显式声明空的输出结构。

def add_intraday_range(price_partition):  # 定义保持行数不变的分区函数
    '''根据真实高低价和收盘价添加相对日内振幅。'''
    return price_partition.assign(intraday_range=(price_partition['high'] - price_partition['low']) / price_partition['adj_close'])  # 用向量化Pandas运算生成新列

range_meta = dask_yrd._meta.assign(intraday_range=pd.Series(dtype='float64'))  # 声明空输出的列名与类型
dask_with_range = dask_yrd.map_partitions(add_intraday_range, meta=range_meta)  # 在不试跑真实分区的情况下构建任务图
range_summary_lazy = dask_with_range.groupby('order_book_id')['intraday_range'].mean()  # 继续构建证券级平均振幅聚合
range_summary = range_summary_lazy.compute().sort_values(ascending=False)  # 只物化四行统计结果
range_summary.to_frame('mean_intraday_range')  # 输出真实行情结果
mean_intraday_range
order_book_id
002415.XSHE 0.030899
600276.XSHG 0.026979
002142.XSHE 0.025389
600104.XSHG 0.015661

错误用法对照：返回 schema 随数据变化

# 该片段仅展示反模式，不在教材中执行
def unstable_schema(partition):  # 定义输出列依赖分区内容的错误函数
    if partition['daily_return'].mean() > 0:  # 不同分区可能进入不同分支
        return partition.assign(regime='positive')  # 该分支增加regime列
    return partition  # 另一分支不增加列，违反统一meta契约

所有分区函数都必须返回相同列和兼容 dtype；业务分类应成为一列的值，而不是改变 schema。

14.9 .compute()、.persist() 与内存边界

.compute() 执行所需图并把最终集合转换为本地对象；.persist() 执行并让分区继续以 Dask 集合形式驻留在 worker 内存中。二者都可能触发计算，但物化位置不同。

安全决策应依次问：

  1. 最终结果有多少行、多少列、何种 dtype？
  2. 最大中间算子是否需要 shuffle、排序或连接状态？
  3. worker 是否有 spill 空间，驱动进程是否能容纳最终本地对象？
  4. 是否可先聚合、抽样、head()，或直接用 to_parquet() 分区写出？

若估计每行结果平均占 \(b\) 字节，返回 \(n_r\) 行，则驱动端至少需要约 \(n_rb\)，还未计对象开销与转换峰值。compute() 不是数据量检查函数；应在调用前缩小结果。

safe_result_lazy = dask_yrd.groupby('order_book_id').agg({'daily_return': ['count', 'mean', 'std'], 'total_turnover': 'sum'})  # 将日度明细压缩为证券级充分统计量
expected_group_count = len(YRD_CODES)  # 根据业务样本预先界定输出上限
safe_result = safe_result_lazy.compute()  # 只将小型聚合结果传回驱动进程
assert len(safe_result) <= expected_group_count  # 验证实际结果未超过预期证券数
safe_result  # 展示可安全物化的真实统计结果
daily_return total_turnover
count mean std sum
order_book_id
002415.XSHE 241 0.000346 0.020718 3.286586e+11
600104.XSHG 241 -0.000106 0.010892 6.962311e+10
600276.XSHG 241 0.000869 0.019035 3.781830e+11
002142.XSHE 241 -0.001709 0.017976 1.993612e+11

14.10 常见误区

  1. 认为 API 相似就能逐行迁移 Pandas。 .iloc、隐式索引、行顺序和许多全局操作的语义不同。
  2. 把分区当成随机小样本。 分区由文件与索引布局决定，可能存在时间或证券聚集与严重倾斜。
  3. 频繁调用 .compute()。 每次可能重复执行上游图；若确需复用且 worker 内存足够，可有选择地 persist()。
  4. 忽略 shuffle。 它常是网络、磁盘与峰值内存的主导成本。
  5. 省略 meta。 推断可能执行函数并得到错误 schema，尤其不适用于有副作用或数据依赖分支的 UDF。
  6. 认为 distributed 会让错误算法自动扩展。 巨大单分区、倾斜键和驱动端全量 compute() 在集群上仍会失败。

14.11 本章小结

Dask 的基本单位是任务和分区。任务图揭示依赖与可并行性，divisions 为有序索引操作提供边界，shuffle 则为跨分区共置数据付出代价。调度器必须匹配 GIL 与数据局部性，meta 必须稳定描述输出，最终是否安全取决于 worker 中间状态和驱动端物化结果两个内存边界。

14.12 分层练习与完整答案

14.12.1 练习 14.1：任务图与延迟执行

为什么打印一个 Dask Series 不会得到完整聚合值？如果任务图有总工作量 80 秒、关键路径 18 秒，使用 8 个理想 worker 的运行时间下界是多少？

答案

打印延迟集合只展示结构，因为业务任务尚未由终端动作触发。由 式 14.1，\(T_8\geq\max(80/8,18)=18\) 秒；关键路径而非平均工作量成为下界。

14.12.2 练习 14.2：分区与 divisions

验证日期索引对象对 2023 年第三季度的切片结果是否全部落在指定区间，并说明已知 divisions 与唯一索引的区别。

答案

third_quarter = date_indexed.loc['2023-07-01':'2023-09-30'][['order_book_id', 'adj_close']].compute()  # 利用日期divisions物化第三季度目标列
third_quarter_dates = third_quarter.index.to_series()  # 取得物化结果的日期索引
assert third_quarter_dates.between('2023-07-01', '2023-09-30').all()  # 验证边界裁剪语义
print(third_quarter.groupby('order_book_id')['adj_close'].count())  # 报告各证券季度交易日数量
order_book_id
002142.XSHE    64
002415.XSHE    64
600104.XSHG    64
600276.XSHG    64
Name: adj_close, dtype: int64

已知 divisions 只表示分区索引范围已知且有序；同一日期可以对应多家公司，因此索引不必唯一。

14.12.3 练习 14.3：meta 与真实数据

用 map_partitions() 新增成交额亿元列，显式提供 meta，再计算每家公司日均成交额亿元。

答案

def add_turnover_billion(price_partition):  # 定义保持原行粒度的金额单位转换
    '''把成交额从元转换为亿元并保持固定schema。'''
    return price_partition.assign(turnover_billion=price_partition['total_turnover'] / 1e8)  # 使用向量化除法转换金融单位

turnover_meta = dask_yrd._meta.assign(turnover_billion=pd.Series(dtype='float64'))  # 声明新增列的精确输出类型
dask_turnover_billion = dask_yrd.map_partitions(add_turnover_billion, meta=turnover_meta)  # 构建带稳定schema的分区任务
mean_turnover_lazy = dask_turnover_billion.groupby('order_book_id')['turnover_billion'].mean()  # 延迟计算证券日均成交额
mean_turnover = mean_turnover_lazy.compute().sort_values(ascending=False)  # 只把四行结果带回驱动进程
mean_turnover.to_frame('mean_daily_turnover_billion')  # 输出单位明确的真实结果
mean_daily_turnover_billion
order_book_id
600276.XSHG 15.692242
002415.XSHE 13.637288
002142.XSHE 8.272249
600104.XSHG 2.888926

14.12.4 练习 14.4：综合 shuffle 与结果核验

用 Dask 计算公司月度收益率标准差，设置两个输出分区；再用 Pandas 对已经筛选的小样本核验结果。解释该任务为何可能需要 shuffle。

答案

monthly_returns = dask_yrd[['order_book_id', 'trade_date', 'daily_return']].assign(trade_month=dask_yrd['trade_date'].dt.to_period('M').astype(str))  # 在窄表上生成公司月键
monthly_volatility_lazy = monthly_returns.groupby(['order_book_id', 'trade_month'])['daily_return'].std(split_out=2)  # 将组结果哈希分配到两个输出分区
monthly_volatility_dask = monthly_volatility_lazy.compute().sort_index()  # 物化约四十八个公司月结果
monthly_volatility_dask.head(12)  # 展示月度波动率序列
order_book_id  trade_month
002142.XSHE    2023-01        0.018031
               2023-02        0.016777
               2023-03        0.015388
               2023-04        0.022311
               2023-05        0.017667
               2023-06        0.015215
               2023-07        0.025005
               2023-08        0.015534
               2023-09        0.019300
               2023-10        0.018977
               2023-11        0.010563
               2023-12        0.017663
Name: daily_return, dtype: float64
monthly_returns_pandas = dask_yrd[['order_book_id', 'trade_date', 'daily_return']].compute()  # 仅为教学核验物化四家公司小样本
monthly_returns_pandas['trade_month'] = monthly_returns_pandas['trade_date'].dt.to_period('M').astype(str)  # 在Pandas中构造相同月份键
monthly_volatility_pandas = monthly_returns_pandas.groupby(['order_book_id', 'trade_month'])['daily_return'].std().sort_index()  # 计算同口径基准结果
pd.testing.assert_series_equal(monthly_volatility_dask, monthly_volatility_pandas, check_names=False)  # 验证Dask分区计算与Pandas一致
print(monthly_volatility_pandas.head(12))  # 输出已核验的部分结果
order_book_id  trade_month
002142.XSHE    2023-01        0.018031
               2023-02        0.016777
               2023-03        0.015388
               2023-04        0.022311
               2023-05        0.017667
               2023-06        0.015215
               2023-07        0.025005
               2023-08        0.015534
               2023-09        0.019300
               2023-10        0.018977
               2023-11        0.010563
               2023-12        0.017663
Name: daily_return, dtype: float64

同一公司月份的日记录可能分散在多个输入文件分区；两个输出分区要求按组合键重新路由或合并部分统计量，因此会出现 hash shuffle 或等价的跨分区通信。

14.12.5 练习 14.5：调度器决策

分别为以下任务选择首选调度器并说明需要实测的反例条件：（1）本机 Parquet 解码加 NumPy 聚合；（2）不释放 GIL 的纯 Python 文本 UDF，且分区可序列化；（3）跨三台机器、需要仪表板和 worker 失败恢复的长任务。答案必须同时讨论 GIL、序列化或网络，以及数据局部性，不能只写调度器名称。

答案

（1）首选 threads：Parquet 与 NumPy 的主要算子通常在原生代码中释放 GIL，共享地址空间还避免复制大分区；若任务很小，调度开销可能使 synchronous 更快，必须同口径实测。（2）可从 processes 起步：独立解释器能绕过 GIL，但只有当纯 Python 计算收益超过分区序列化、进程启动和内存复制代价时才成立。（3）选择 distributed：它提供跨机调度、监控与容错，但应让数据靠近 worker，并审计网络传输、spill 和 worker 内存；把本地分片反复跨网搬运可能抵消并行收益。