我正在使用 AWS Glue 将多个文件从 S3 移动到 RDS 实例。每天我都会在 S3 中获取一个新文件，该文件可能包含新数据，但也可能包含我已经保存的带有一些更新值的记录。如果我多次运行该作业，我当然会在数据库中获得重复的记录。如果 Glue 注意到字段已更改，我希望 Glue 尝试更新该记录，而不是插入多条记录，每条记录都有一个唯一的 ID。这可能吗？
【问题讨论】：
标签： python amazon-web-services pyspark etl aws-glue
我采用了 Yuriy 建议的第二个选项的类似方法。获取现有数据以及新数据，然后进行一些处理以合并到它们并以覆盖模式写入。以下代码将帮助您了解如何解决此问题。
sc = SparkContext()
glueContext = GlueContext(sc)
#get your source data
src_data = create_dynamic_frame.from_catalog(database = src_db, table_name = src_tbl)
src_df = src_data.toDF()
#get your destination data
dst_data = create_dynamic_frame.from_catalog(database = dst_db, table_name = dst_tbl)
dst_df = dst_data.toDF()
#Now merge two data frames to remove duplicates
merged_df = dst_df.union(src_df)
#Finally save data to destination with OVERWRITE mode
merged_df.write.format('jdbc').options( url = dest_jdbc_url,
user = dest_user_name,
password = dest_password,
dbtable = dest_tbl ).mode("overwrite").save()
【讨论】：
不幸的是，使用 Glue 并没有优雅的方法来做到这一点。如果你想写信给 Redshift，你可以use postactions to implement Redshift merge 操作。但是，其他 jdbc 接收器 (afaik) 是不可能的。
或者，在您的 ETL 脚本中，您可以从数据库加载现有数据以在保存之前过滤掉现有记录。但是，如果您的数据库表很大，那么该作业可能需要一段时间来处理它。
另一种方法是先以“覆盖”模式写入暂存表（替换现有暂存数据），然后通过 API 调用数据库以仅将新记录复制到最终表中。
【讨论】：
我已经在表中使用了 INSERT .... ON DUPLICATE KEY.. 用于将 UPSERT 插入到运行 mysql 引擎的 Aurora RDS 中。也许这将是您用例的参考。我们不能使用 JDBC，因为我们目前只支持 APPEND、OVERWRITE、ERROR 模式。
我不确定您使用的 RDS 数据库引擎，以下是 mysql UPSERTS 的示例。
请参阅此参考资料，我在其中发布了一个使用 INSERT INTO TABLE..ON DUPLICATE KEY for mysql 的解决方案：
Error while using INSERT INTO table ON DUPLICATE KEY, using a for loop array
【讨论】：