如何去除Lake Formation托管表中的重复数据？

针对同一用户修改地址后产生重复记录、仅需保留最新条的需求，可通过以下几种方案实现：

方案一：Glue作业内通过Spark逻辑实时去重 #

这是最直接的处理方式，核心依赖用户唯一标识和更新时间戳两个字段：

• 确保每条Kinesis记录携带用户唯一ID（如user_id）和准确的更新时间戳（如update_ts，建议用用户操作发生的时间而非流接收时间）
• 在Glue作业中，将读取的Kinesis数据转换为Spark DataFrame后，使用窗口函数分组去重：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col

# 假设读取Kinesis后的DataFrame为raw_df
window_spec = Window.partitionBy("user_id").orderBy(col("update_ts").desc())
deduped_df = raw_df.withColumn("row_rank", row_number().over(window_spec)) \
                   .filter(col("row_rank") == 1) \
                   .drop("row_rank")

• 将去重后的DataFrame转换回DynamicFrame，写入Lake Formation托管表即可

方案二：利用Lake Formation事务型表的Merge操作 #

如果你的Lake Formation托管表支持事务（需为Parquet/ORC格式且开启事务），可以用Merge操作实现增量更新时的自动去重：

• 先确保表已开启事务，在Glue作业中执行Merge逻辑，匹配相同user_id的记录，仅保留更新时间戳更大的条目：

from awsglue.dynamicframe import DynamicFrame

# 将去重后的DataFrame转为DynamicFrame
deduped_dyf = DynamicFrame.fromDF(deduped_df, glueContext, "deduped_dyf")

# 执行Merge，替换旧记录或插入新记录
glueContext.merge_dynamic_frame(
    frame=deduped_dyf,
    database="your_database_name",
    table_name="your_table_name",
    transformation_ctx="merge_transformation",
    merge_conditions=[
        "source.user_id = target.user_id",
        "source.update_ts > target.update_ts"
    ],
    update_condition="source.update_ts > target.update_ts",
    insert_condition="source.user_id NOT IN (SELECT user_id FROM target)"
)

注意：Merge操作需要Glue版本支持，且作业需有对应的Lake Formation权限。

方案三：从源头减少重复数据 #

如果可控，可在数据流入Kinesis前做预处理：

• 调整数据生产者逻辑，同一用户的连续更新仅发送最新记录到Kinesis流
• 用Kinesis Data Analytics对流入数据做实时去重，再输出给Glue任务处理

额外注意事项 #

• 若表中已有历史重复数据，可单独编写Glue批处理任务，全量读取表数据后用窗口函数去重，再覆盖回原表（操作前务必备份数据）
• 数据量较大时，窗口函数去重可配合分区（按user_id或时间分区）优化性能
• 确保update_ts字段的准确性，避免因时间误差导致去重逻辑失效

内容的提问来源于stack exchange，提问作者kylie.zoltan