调用dask.map_partitions().compute()返回2GB结果时16GB本地机器报内存错误

你触发的是本地msgpack序列化阶段的内存错误，和集群内存无关，核心原因有两个：

• 你预期的2GB返回结果是逻辑体积，msgpack在序列化pandas DataFrame进行网络传输时，需要申请2_{3倍于数据本身大小的临时内存做格式转换，即使实际返回数据只有2GB，序列化阶段也需要至少4}6GB的额外临时空间。你的16GB本地内存还要分摊系统、其他应用、Python进程本身的占用，剩余可用内存不足以支撑序列化开销就会触发MemoryError
• 你的实际返回结果可能远大于预期的2GB：如果你的DataFrame包含大量object类型的字符串列，Dask的默认内存预估会严重偏小，实际返回体积可能达到预估的2~4倍，进一步放大序列化开销

先不要直接拉取全量结果，先执行以下代码确认实际返回数据的真实体积：

# 计算Dask DataFrame的实际总内存占用（单位：字节）
total_size = (read_parquets_separately_by_dask_and_concatenate('hub_to_hub_capacity/2021/10/')
    .map_partitions(capacity.capacity_features,
                    meta=meta_capacity_features,
                    transform_divisions=False)
    .memory_usage(deep=True).sum().compute())
print(f"实际返回数据总大小：{total_size / 1024**3:.2f} GB")

• 优先方案：不要直接全量拉取结果到本地，先将计算后的结果写入集群共享存储（比如Azure Blob、集群节点可访问的分布式文件系统）的parquet文件，后续再分批次下载读取到本地，完全规避一次性序列化传输的高内存开销
• 必须拉取到本地的场景：将结果拆分为单分区逐次拉取，避免全量数据同时进入序列化流程：

import pandas as pd
from dask.base import to_delayed

dask_client.upload_file(os.path.join(src_folder,'capacity.py'))
dask_df = (read_parquets_separately_by_dask_and_concatenate('hub_to_hub_capacity/2021/10/')
    .map_partitions(capacity.capacity_features,
                    meta=meta_capacity_features,
                    transform_divisions=False)
)
# 拆分为分区级延迟对象，逐块拉取到本地
part_delayed = to_delayed(dask_df, optimize_graph=True)
local_parts = []
for part in part_delayed:
    local_parts.append(dask_client.compute(part).result())
# 本地拼接全量结果
result = pd.concat(local_parts, ignore_index=True)

• 优化返回结果体积：在capacity_features函数中将object类型列转成category类型，数值列改用更小精度的dtype（比如int64转int32、float64转float32，只要满足业务精度要求即可），可以将返回体积压缩30%~70%，大幅降低序列化开销
• 系统层面优化：如果是Windows系统，手动调大系统虚拟内存/页面文件大小，也可以缓解内存不足的问题

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