如何在Scala的AWS Glue中用applyMapping处理数组内对象字段？

AWS Glue原生的applyMapping确实不支持直接访问数组内的嵌套字段进行映射，但可以通过自定义转换函数+map算子的方式，实现类似applyMapping的简洁字段映射逻辑，避免relationalize多步操作的繁琐。

核心思路 #

将字段映射规则以类似applyMapping的列表形式定义，然后通过自定义函数遍历处理数组内的每个对象，完成字段重命名、删除、嵌套结构调整，最后用map算子批量应用到整个DynamicFrame。

具体实现代码 #

from awsglue.transforms import *
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, BooleanType, ArrayType

# 定义映射规则，格式：(原字段路径, 目标字段名, 字段类型)
# 顶层字段映射
top_level_mappings = [
    ("uuid", "uuid", StringType()),
    ("item_id", "item_code", StringType()),
    # 直接忽略net_price字段，实现删除
]

# coupons数组内对象的映射规则
coupon_mappings = [
    ("code", "coupon_code", StringType()),
    ("type", "type", StringType()),
    ("amount", "amount", IntegerType()),
    # 忽略valid和meta字段
]

# conditions数组内对象的映射规则
condition_mappings = [
    ("day", "days_valid", ArrayType(StringType())),
    ("time_start", "time_start", StringType()),
    ("time_end", "time_end", StringType()),
    # 忽略bank字段
]

# 自定义转换函数，处理单条记录
def transform_record(record):
    # 处理顶层字段
    transformed = {}
    for src, dst, _ in top_level_mappings:
        transformed[dst] = record.get(src)
    
    # 处理coupons数组
    if "coupons" in record and record["coupons"]:
        transformed_coupons = []
        for coupon in record["coupons"]:
            coupon_obj = {}
            # 映射coupon字段
            for src, dst, _ in coupon_mappings:
                coupon_obj[dst] = coupon.get(src)
            
            # 处理coupons内的conditions数组
            if "conditions" in coupon and coupon["conditions"]:
                transformed_conditions = []
                for cond in coupon["conditions"]:
                    cond_obj = {}
                    for src, dst, _ in condition_mappings:
                        cond_obj[dst] = cond.get(src)
                    transformed_conditions.append(cond_obj)
                coupon_obj["conditions"] = transformed_conditions
            transformed_coupons.append(coupon_obj)
        transformed["coupons"] = transformed_coupons
    
    return transformed

# 假设原始DynamicFrame为original_dyf
processed_dyf = original_dyf.map(f=transform_record)

代码说明 #

1. 映射规则定义：完全模仿applyMapping的字段映射格式，清晰直观，便于维护。
2. 自定义转换函数：递归处理嵌套数组，按照映射规则完成字段重命名和冗余字段删除，逻辑完全可控。
3. map算子应用：直接对DynamicFrame进行批量转换，无需拆分表再合并，比relationalize流程更简洁。

这种方式既保留了applyMapping的便捷性（通过映射规则管理字段），又解决了嵌套数组的处理问题，代码可读性和维护性都很强。

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