Dask pivot_table运算极慢、易触发KilledWorker异常问题咨询

• 业务目标：构建透视表统计不同日期维度下各基金对应的持仓头寸信息。数据源为按天存储的大体积parquet文件，读取时仅提取as_of_date、position_id、fund3个字段；1年跨度数据经透视后结果约为10万行、365列，单元格取值为NaN或短字符串格式的基金名称。
• 对照方案1：将Dask DataFrame调用compute()转为pandas对象后再执行透视，1年数据处理总耗时46秒，实现代码：

ddf = dd.read_parquet("s3://.../2021*.parquet", columns=["as_of_date", "position_id", "fund"]).categorize(columns=['as_of_date'])
ddf.compute().pivot_table(index='position_id', columns='as_of_date', values='fund', aggfunc='first')

该方案缺陷是会产生对象存储到本地的数据传输开销。

• 对照方案2：直接调用Dask内置pivot_table接口执行透视再compute()，原本预期效率不低于方案1，实际测试1年数据处理耗时达12分钟；数据跨度超过1年时任务直接失败，抛出KilledWorker异常。实现代码：

ddf = dd.read_parquet("s3://.../2021*.parquet", columns=["as_of_date", "position_id", "fund"]).categorize(columns=['as_of_date'])
ddf.pivot_table(index='position_id', columns='as_of_date', values='fund', aggfunc='first').compute()

• 已知前提：源数据不存在重复值，因Dask未提供无聚合的pivot()接口，使用aggfunc='first'作为最简聚合参数实现透视逻辑，需要定位Dask原生pivot_table性能极差的根本原因。

• Shuffle开销量级差异：Daskpivot_table的底层实现是先按index+columns组合键做分组聚合，再执行宽表转换。该过程需要将所有相同position_id的数据跨分区shuffle到同一节点，当前场景下分组组合键基数达10万*365=3650万级别，shuffle产生的中间数据落盘、网络传输、分区排序开销远高于直接拉取窄表到本地pandas计算的传输开销。
• 分类列广播与空值填充开销：提前对as_of_date做categorize操作后，Dask会将全量日期分类元数据广播到所有工作分区，每个分区需要预先创建全部365个结果列做空值填充，随着数据跨度增长、分区数增加，这部分内存和计算开销会线性上涨，最终触发Worker内存溢出被强制杀死，抛出KilledWorker异常。
• 聚合逻辑无捷径优化：虽然源数据无重复键、first聚合理论上不需要做值比对，但Dask分组聚合模块不会识别该特殊场景，依然会走完整的分组建索引、值遍历选取流程，产生大量无意义计算开销。

• 当前1年跨度透视结果总内存占用不足1GB，完全可以放入单机内存，直接使用compute()转pandas后透视就是最优方案，无需强行使用Dask原生接口，分布式计算的shuffle开销在这种小结果集的宽表透视场景下远高于单机计算的传输开销。
• 若后续数据量增长到单机内存无法承载，不要直接调用Daskpivot_table，可按as_of_date逐批次拉取数据做单天维度的局部透视，再按position_id分批次做外连接拼接，避免全量数据shuffle。

内容的提问来源于stack exchange，提问作者lunix