AWS Glue使用ResolveChoice投射timestamp时字段丢失问题

我之前也碰到过这个Glue转换的坑——用ResolveChoice的project:timestamp处理long类型时间戳时，字段莫名丢失。结合你的代码和问题描述，我来拆解下原因和解决办法：

为什么timestamp字段会丢失？ #

首先，ResolveChoice的project操作是用来解决类型歧义的（比如同一列同时存在string和int两种类型），它的逻辑是把所有数据投射成一种统一类型。但你的场景里，timestamp字段在ApplyMapping后已经是明确的long类型，没有歧义需要解决，这时候强行用project:timestamp，Glue的类型解析逻辑会认为这个字段不符合投射要求，直接把它剔除了。

另外，project:timestamp对long类型的支持并不好，它更适合处理本身就是时间格式字符串或混合类型的字段，而不是从long转timestamp的场景。

可行的解决方案 #

方案1：直接在ApplyMapping中转换类型 #

如果原始的timestamp是字符串格式的时间戳（比如毫秒/秒数），可以跳过中间转long的步骤，直接在ApplyMapping里把它转成timestamp类型：

datasource0 = ApplyMapping.apply(
    frame = datasource0, 
    mappings = [
        ("timestamp", "string", "timestamp", "timestamp"),  # 直接从string映射到timestamp
        ("name", "string", "name", "string"), 
        ("value", "string", "value", "string"), 
        ("type", "string", "type", "string")
    ]
)

如果原始字符串是毫秒级时间戳，需要先转long再转timestamp的话，可以用自定义映射函数：

def parse_timestamp(rec):
    # 把字符串转成long，再转成timestamp（毫秒转秒除以1000）
    rec["timestamp"] = int(rec["timestamp"]) / 1000
    return rec

datasource0 = Map.apply(frame=datasource0, f=parse_timestamp)
datasource0 = ApplyMapping.apply(
    frame = datasource0, 
    mappings = [
        ("timestamp", "double", "timestamp", "timestamp"),
        ("name", "string", "name", "string"), 
        ("value", "string", "value", "string"), 
        ("type", "string", "type", "string")
    ]
)

方案2：用ResolveChoice的cast:timestamp替代project:timestamp #

既然你的字段类型是明确的long，不需要解决歧义，直接用cast操作来强制转换类型就好，这比project更适合纯类型转换场景：

datasource0 = ResolveChoice.apply(
    frame = datasource0, 
    specs = [
        ('timestamp','cast:timestamp'),  # 替换成cast操作
        ('name','cast:string'), 
        ('type','cast:string'), 
        ('value','cast:string')
    ]
)

方案3：转成Spark DataFrame处理（更灵活） #

Glue Dynamic Frame的类型转换有时候不如原生Spark灵活，你可以先转成DataFrame，用Spark的函数处理后再转回Dynamic Frame：

from pyspark.sql.functions import from_unixtime

# 把Dynamic Frame转成Spark DataFrame
df = datasource0.toDF()
# 将long类型的时间戳（假设是毫秒级）转成timestamp
df = df.withColumn("timestamp", from_unixtime(df["timestamp"] / 1000).cast("timestamp"))
# 转回Glue Dynamic Frame
datasource0 = glueContext.create_dynamic_frame.from_df(df, glueContext, "timestamp_conversion")

验证步骤 #

在写入Parquet之前，建议先打印下Schema确认：

datasource0.printSchema()

如果输出里能看到timestamp字段类型为timestamp，再执行写入操作就没问题了。

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