Amazon EMR PySpark作业迁移至SageMaker时报dict_keys无法pickle错误

这个报错的核心是Python2和Python3对dict.keys()方法的返回值类型差异，以及PySpark的序列化机制共同导致的：

• Python 2环境中，dict.keys()返回值是普通列表类型，原生支持pickle序列化
• Python 3环境中，dict.keys()返回的是dict_keys视图对象，该对象不支持pickle序列化
• PySpark执行RDD算子时，会将算子函数闭包引用的所有外部变量序列化后分发到Executor节点执行，你代码中return_row函数引用的外部变量（大概率是kpi_list）是直接取某字典的keys()返回值，未做类型转换，属于dict_keys类型，序列化时触发报错
• 报错出现在collect()操作时是因为Spark是懒执行引擎，只有触发action操作才会实际提交作业、执行序列化逻辑

你只需要把所有在RDD算子函数中引用的、来自dict.keys()的变量显式转为列表即可：

• 如果kpi_list是通过xxx_dict.keys()赋值的，修改为 kpi_list = list(xxx_dict.keys())
• 其他所有在return_row、lambda等算子函数中引用的字典键值变量，都统一用list()转换为普通列表
• 可选优化：对于cols_dict这类只读的公共字典变量，可以用sc.broadcast(cols_dict)广播到所有节点，减少重复序列化开销，修改示例如下：

# 广播公共字典示例
broadcast_cols = sc.broadcast(cols_dict)
def return_row(z):
    cols_dict = broadcast_cols.value
    # 原有逻辑保持不变
    temp = z.split(delimiter)
    row = [float(temp[cols_dict[kpi]]) if temp[cols_dict[kpi]] != '' else 0.0 for kpi in (kpi_list + other_kpis)] + [
        temp[cols_dict['eventids']]] + [temp[cols_dict['conversions']]]
    return row

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