如何在AWS Glue DynamicFrame多列中批量修改值且不转换为DataFrame

你遇到的PicklingError是因为直接将类实例方法传入了Map转换。Glue的Map转换在分布式执行时需要将传入的函数序列化后分发到Executor节点，而类实例方法会绑定整个类对象，如果类中包含GlueContext、SparkSession这类无法被序列化的Py4J Java包装对象，就会触发序列化失败。
另外你提供的YAML配置存在语法错误，column_2下的映射列表缺少values字段声明，需要先修正为合法格式：

value_mapping:
    column_1:
        column_name: asd
        values:
            - old_value_1: new_value_1
            - old_value_2: new_value_2
    column_2:
        column_name: dsa
        values:
            - old_value_1: new_value_1
            - old_value_2: new_value_2

方案1：调整Map函数实现，避免序列化整个类实例 #

核心逻辑是将映射规则预处理为独立的可序列化字典，映射函数不绑定类实例，仅引用预处理后的规则字典即可。

步骤1：预处理映射规则 #

在类初始化阶段提前处理配置，生成扁平的可序列化映射结构：

def __init__(self, config):
    self.config = config
    # 预处理映射规则，结构为 {列名: {旧值: 新值}}
    self.serializable_mapping = {}
    for item in config['value_mapping'].values():
        col = item['column_name']
        val_map = {}
        for pair in item['values']:
            val_map.update(pair)
        self.serializable_mapping[col] = val_map

步骤2：改写Map调用逻辑 #

映射函数使用闭包实现，仅引用独立的映射规则字典，不绑定self：

def map_values_in_columns(self, df):
    mapping = self.serializable_mapping
    def _map_func(rec):
        for col_name, val_map in mapping.items():
            if rec[col_name] in val_map:
                rec[col_name] = val_map[rec[col_name]]
        return rec
    return Map.apply(frame = df, f = _map_func)

方案2：使用SqlQuery转换（性能更优） #

直接通过Glue内置的SqlQuery转换对DynamicFrame执行SQL操作，无需自定义Python函数，避免序列化开销，适合大数据量场景：

from awsglue.transforms import SqlQuery

def map_values_in_columns(self, df):
    all_cols = df.columns()
    select_parts = []
    # 拼接需要替换的列的CASE语句
    for col_name, val_map in self.serializable_mapping.items():
        when_clause = " ".join([f"WHEN {col_name} = '{old}' THEN '{new}'" for old, new in val_map.items()])
        case_stmt = f"CASE {when_clause} ELSE {col_name} END AS {col_name}"
        select_parts.append(case_stmt)
    # 拼接不需要替换的列
    for col in all_cols:
        if col not in self.serializable_mapping:
            select_parts.append(col)
    select_clause = ", ".join(select_parts)
    return SqlQuery.apply(
        frame = df,
        query = f"SELECT {select_clause} FROM temp_table",
        name = "value_replace",
        mapping = {"temp_table": df}
    )

内容的提问来源于stack exchange，提问作者E. Faslo