胶水创建重复记录，如何解决？

Posted 2023-03-31

【中文标题】胶水创建重复记录，如何解决？

【英文标题】：Glue creates duplicates of records, how to fix it?

【发布时间】：2019-03-08 09:01:14

【问题描述】：

目前，我们使用 Glue（python 脚本）将数据从 mysql 数据库迁移到 RedShift 数据库。 昨天，我们发现一个问题：有些记录是重复的，这些记录具有相同的主键，在 MySQL 数据库中使用。根据我们的要求，RedShift 数据库中的所有数据都应该与 MySQL 数据库中的数据相同。

我试图在迁移之前删除一个 RedShift 表，但没有找到方法...

你能帮我解决这个问题吗？

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [TempDir, JOB_NAME]
args = getResolvedOptions(sys.argv, ['TempDir','JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "glue-db", table_name = "table", transformation_ctx = "datasource0")
applymapping0_1 = ApplyMapping.apply(frame = datasource0, mappings = [...], transformation_ctx = "applymapping0_1")
resolvechoice0_2 = ResolveChoice.apply(frame = applymapping0_1, choice = "make_cols", transformation_ctx = "resolvechoice0_2")
dropnullfields0_3 = DropNullFields.apply(frame = resolvechoice0_2, transformation_ctx = "dropnullfields0_3")
datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = "dbtable": "table", "database": "database", redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink0_4")

我的解决方案是：

datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = "dbtable": "mytable", "database": "mydatabase", "preactions": "delete from public.mytable;"

【参考方案1】：

如果您的目标是在目标表中不存在重复项，您可以为 JBDC 接收器使用 postactions 选项（有关详细信息，请参阅 this answer）。基本上它允许使用临时表实现Redshift merge。

对于您的情况，它应该是这样的（替换现有记录）：

post_actions = (
         "DELETE FROM dest_table USING staging_table AS S WHERE dest_table.id = S.id;"
         "INSERT INTO dest_table (id,name) SELECT id,name FROM staging_table;"
         "DROP TABLE IF EXISTS staging_table"
    )
datasink0_4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields0_3, catalog_connection = "redshift-cluster", connection_options = "dbtable": "staging_table", "database": "database", "overwrite" -> "true", "postactions" -> post_actions, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink0_4")

【讨论】：

尤里·邦达鲁克，感谢您的回复！我应用了类似的解决方案。 "preactions": "从 public.mytable 中删除；"

@Luzifer 太棒了！很高兴它帮助您解决了您的问题！您能否将答案标记为已接受？

【参考方案2】：

Redshift 不施加唯一键约束

除非您可以保证您的源脚本避免重复，否则您需要运行常规作业以在 redshift 上进行重复数据删除，

delete from yourtable
where id in
(
select id
from yourtable
group by 1
having count(*) >1
)
;

您是否认为 DMS 可以替代 Glue？这可能更适合您。

【讨论】：

感谢您的回复！目前，我们使用胶水，并且在不久的将来我们无法更改它......也许你知道，在作业脚本中（我已在主帖中添加），我可以输入类似“删除表 table_name;”的查询哪个会在迁移开始之前从 redshift 中删除迁移表？

我认为你不能轻易做到这一点。见***.com/questions/46228253/…