AWS Glue 到 Redshift：重复数据？答案

【问题标题】：AWS Glue to Redshift: duplicate data?AWS Glue 到 Redshift：重复数据？
【发布时间】：2019-02-23 03:26:42
【问题描述】：

以下是关于我如何设置的一些要点：

我已将 CSV 文件上传到 S3，并设置了 Glue 爬虫来创建表和架构。我有一个 Glue 作业设置，它使用 JDBC 连接将 Glue 表中的数据写入我们的 Amazon Redshift 数据库。 Job 还负责映射列和创建红移表。通过重新运行作业，我在 redshift 中得到了重复的行（如预期的那样）。

但是，有没有办法在插入新数据之前替换或删除行？

BOOKMARK 功能已启用但无法正常工作。

如何连接到 redshift，在将数据推送到 Python 中的 redshift 之前删除所有数据作为 JOB 的一部分？

【问题讨论】：

标签： python amazon-web-services amazon-s3 amazon-redshift aws-glue

【解决方案1】：

目前 Glue 不支持为 JDBC 源添加书签。

您可以使用 postactions 选项（Scala 中的代码）在 Glue 作业中实现 upsert/merge into Redshift：

val fields = sourceDf.columns.mkString(",")

glueContext.getJDBCSink(
  catalogConnection = "RedshiftConnectionTest",
  options = JsonOptions(Map(
    "database" -> "conndb",
    "dbtable" -> "staging_schema.staging_table",
    "postactions" -> 
        s"""
           DELETE FROM dst_schema.dst_table USING staging_schema.staging_table AS S WHERE dst_table.id = S.id;
           INSERT INTO dst_schema.dst_table ($fields) SELECT $fields FROM staging_schema.staging_table;
           DROP TABLE IF EXISTS staging_schema.staging_table
        """
  )),
  redshiftTmpDir = tempDir,
  transformationContext = "redshift-output"
).writeDynamicFrame(DynamicFrame(sourceDf, glueContext))

如果你只是想删除现有的表，那么你可以使用preactions参数来代替：

glueContext.getJDBCSink(
  catalogConnection = "RedshiftConnectionTest",
  options = JsonOptions(Map(
    "database" -> "conndb",
    "dbtable" -> "dst_schema.dst_table",
    "preactions" -> "DELETE FROM dst_schema.dst_table"
  )),
  redshiftTmpDir = tempDir,
  transformationContext = "redshift-output"
).writeDynamicFrame(DynamicFrame(sourceDf, glueContext))

【讨论】：

【解决方案2】：

只要您的表上有唯一键，最好是整数主键。那么我解决这个问题的方法如下：

实施调度工具以允许按顺序运行作业。一世推荐 Airflow。
启动 Glue 作业以从源读取并写入暂存桌子。（临时表将仅包含该粘合运行的输出，不一定包含所有行）
等待 Glue 作业完成（使用调度工具）
启动在 Redshift 上运行的 SQL 作业：

a) 从目标表中删除匹配的行

delete from target
where id in (select id from staging);

b) 将数据从 staging 插入到目标表中

insert into target select * from staging;

c) 截断临时表

d) 清理并分析两个表

vacuum target to 100 percent;
analyze target;
vacuum staging;

【讨论】：

【解决方案3】：

您可以使用 python 模块 pg8000 来连接到 Redfshift 并执行 SQL 以从 Glue 脚本中删除（删除/截断）数据。 pg8000 是纯 python，所以它可以与 Glue 一起使用。

查看此链接：AWS Glue - Truncate destination postgres table prior to insert

我试过了，效果很好。希望对你有所帮助，

【讨论】：

【解决方案4】：

如果您希望进行完全加载，可以使用 spark/Pyspark 数据块库来覆盖表：

df.write\
  .format("com.databricks.spark.redshift")\
  .option("url", redshift_url)\
  .option("dbtable", redshift_table)\
  .option("user", user)\
  .option("password", readshift_password)\
  .option("aws_iam_role", redshift_copy_role)\
  .option("tempdir", args["TempDir"])\
  .mode("overwrite")\
  .save()

每个 Databricks/Spark documentation：

覆盖现有表：默认情况下，此库使用执行覆盖的事务，通过删除来实现目标表，创建一个新的空表并追加行给它。

您可以查看 here 中的 databricks 文档

【讨论】：

【解决方案5】：

Glue 作业确实支持使用 JDBC 源添加书签。这一切都取决于是否存在“增加”或“减少”的键（列）。

https://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html

【讨论】：