如何解决AWS Glue中string与struct类型的Choice字段转换问题？

针对你遇到的raw_response字段为Choice类型（string/struct嵌套）的转换需求，由于涉及复杂嵌套结构，不能直接用简单的ResolveChoice指定类型，需要结合嵌套结构遍历和自定义转换逻辑处理，以下是两种可行实现方案：

方法一：ResolveChoice + 自定义UDF（推荐） #

1. 展开嵌套数组 #

先把hooks数组展开，方便单独处理每个元素的response.raw_response字段：

from awsglue.context import GlueContext
from pyspark.sql import SparkSession

# 初始化Glue上下文
glueContext = GlueContext(SparkSession.builder.getOrCreate())

# 读取源数据动态帧
source_frame = glueContext.create_dynamic_frame.from_catalog(
    database="你的数据库名",
    table_name="你的表名"
)

# 展开hooks数组
expanded_frame = source_frame.resolveChoice(specs=[("hooks", "expand")])

2. 定义转换UDF #

根据你提供的struct结构，编写UDF将string类型的raw_response转换为标准struct格式：

from pyspark.sql.functions import udf, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 定义目标raw_response的struct schema
raw_resp_schema = StructType([
    StructField("status_code", IntegerType(), nullable=True),
    StructField("code", StringType(), nullable=True),
    StructField("raw_response", StringType(), nullable=True),
    StructField("message", StringType(), nullable=True),
    StructField("data", StructType([
        StructField("order_id", StringType(), nullable=True),
        StructField("status", StringType(), nullable=True)
    ]), nullable=True),
    StructField("string", StringType(), nullable=True),
    StructField("struct", StructType([
        StructField("data", StructType([
            StructField("order_id", StringType(), nullable=True),
            StructField("status", StringType(), nullable=True)
        ]), nullable=True),
        StructField("status_code", IntegerType(), nullable=True),
        StructField("code", StringType(), nullable=True),
        StructField("raw_response", StringType(), nullable=True),
        StructField("message", StringType(), nullable=True)
    ]), nullable=True)
])

# 转换逻辑：string转struct，struct直接返回
def convert_raw_resp(choice_val):
    if isinstance(choice_val, str):
        return {
            "raw_response": choice_val,
            "status_code": None,
            "code": None,
            "message": None,
            "data": None,
            "string": None,
            "struct": None
        }
    return choice_val

# 注册UDF
convert_udf = udf(convert_raw_resp, raw_resp_schema)

3. 应用转换并重构数组 #

将UDF应用到目标字段后，重新组装hooks数组：

# 转Spark DataFrame处理
df = expanded_frame.toDF()

# 替换response.raw_response字段
processed_df = df.withColumn(
    "response",
    col("response").withField("raw_response", convert_udf(col("response.raw_response")))
)

# 转回动态帧并重新组装数组
final_frame = glueContext.create_dynamic_frame.fromDF(
    processed_df,
    glueContext,
    "final_processed_frame"
).resolveChoice(specs=[("hooks", "array")])

方法二：使用Glue Map转换（无需UDF） #

如果不想编写UDF，可以直接用Glue的Map转换遍历每个记录处理：

from awsglue.transforms import Map

def process_hook_record(rec):
    if "response" in rec and "raw_response" in rec["response"]:
        raw_resp = rec["response"]["raw_response"]
        if isinstance(raw_resp, str):
            # 替换为标准struct结构
            rec["response"]["raw_response"] = {
                "raw_response": raw_resp,
                "status_code": None,
                "code": None,
                "message": None,
                "data": None,
                "string": None,
                "struct": None
            }
    return rec

# 应用Map转换
mapped_frame = Map.apply(frame=source_frame, f=process_hook_record)

# 最终统一将raw_response转为struct类型
final_frame = mapped_frame.resolveChoice(
    specs=[("hooks.element.response.raw_response", "cast:struct")]
)

关键注意事项 #

• 确保目标struct的schema与你提供的完全一致，避免转换后字段缺失或类型不匹配
• 若数据中存在null值，需在转换逻辑中增加null判断，避免报错
• 处理完成后，可通过final_frame.printSchema()检查最终结构是否符合预期

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