使用Glue将S3数据导入DDB失败：重复主键问题解决

问题原因 #

用DISTINCT去重后仍触发主键重复错误，是因为DISTINCT基于整行数据去重——如果两行主键相同但其他字段不同，DISTINCT会保留这两行，写入DynamoDB时就会因为主键重复抛出ValidationException。

修改Glue脚本的解决方案 #

核心思路是仅针对DynamoDB主键字段去重，确保每个主键只保留一条记录（可根据业务逻辑选择保留最新/最早或任意版本的行）。以下是具体修改步骤：

1. 用窗口函数保留最新数据（推荐）

如果数据中有时间戳字段（如update_time），可以按主键分区后取最新的一行：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number
from awsglue.dynamicframe import DynamicFrame

# 原读取Parquet的代码不变
datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/path/"]},
    format="parquet"
)

# 转为Spark DataFrame
df = datasource0.toDF()

# 定义窗口：按主键分区，按时间戳倒序排序
# 若为复合主键，partitionBy需包含所有主键字段，比如partitionBy("user_id", "order_id")
window_spec = Window.partitionBy("your-primary-key").orderBy(df["update_time"].desc())

# 添加行号并过滤出最新行
deduped_df = df.withColumn("row_num", row_number().over(window_spec)) \
               .filter("row_num = 1") \
               .drop("row_num")

# 转回Glue DynamicFrame
deduped_dynamic_frame = DynamicFrame.fromDF(deduped_df, glueContext, "deduped_frame")

# 原写入DynamoDB的代码不变
glueContext.write_dynamic_frame.from_options(
    frame=deduped_dynamic_frame,
    connection_type="dynamodb",
    connection_options={
        "dynamodb.output.tableName": "your-ddb-table-name",
        "dynamodb.throughput.write.percent": "1.0"
    }
)

2. 无时间戳时保留任意一行

如果没有时间戳字段，可通过分组聚合保留每个主键的任意一行：

# 读取并转DataFrame步骤同上
df = datasource0.toDF()

# 按主键分组，对其他字段取第一个值（或用last()取最后一个）
deduped_df = df.groupBy("your-primary-key").agg(
    {col: "first" for col in df.columns if col != "your-primary-key"}
)

# 后续转回DynamicFrame并写入DDB步骤同上

关键注意事项 #

• 如果DynamoDB使用复合主键（分区键+排序键），partitionBy中必须同时包含这两个字段，确保组合键唯一。
• 调整orderBy的字段和排序方向（asc()/desc()），可以控制保留哪一条重复数据。

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