AWS Datapipeline：使用管道将数据从 RDS 数据库（postgres）移动到 Redshift

Question

基本上我正在尝试使用 aws datapipeline 和我正在关注的过程将数据从 postgres 传输到 redshift

1. 编写一个管道（CopyActivity）从postgres to s3移动数据
2. 编写从s3 to redshift 移动数据的管道(RedShiftCopyActivity)

所以就我而言，两者都与我编写的管道完美配合，但问题是数据在 Redshift 数据库中重复

例如下面是来自 postgres 数据库的数据，位于名为 company 的表中

s3 to redshift(RedShiftCopyActivity) 管道成功运行后，数据被复制，但复制如下

以下是 RedShiftCopyActivity(S3 到 Redshift) 管道的一些定义部分

  pipeline_definition = [{
      "id":"redshift_database_instance_output",
      "name":"redshift_database_instance_output",
      "fields":[
          {
          "key" : "database",
          "refValue" : "RedshiftDatabaseId_S34X5",
          },
          {
          "key" : "primaryKeys",
          "stringValue" : "id",
          },
          {
          "key" : "type",
          "stringValue" : "RedshiftDataNode",
          },
          {
          "key" : "tableName",
          "stringValue" : "company",
          },
          {
          "key" : "schedule",
          "refValue" : "DefaultScheduleTime",
          },
          {
          "key" : "schemaName",
          "stringValue" : RedShiftSchemaName,
          },
      ]
  },
  {
      "id":"CopyS3ToRedshift",
      "name":"CopyS3ToRedshift",
      "fields":[
          {
          "key" : "output",
          "refValue" : "redshift_database_instance_output",
          },
          {
          "key" : "input",
          "refValue" : "s3_input_data",
          },
          {
          "key" : "runsOn",
          "refValue" : "ResourceId_z9RNH",
          },
          {
          "key" : "type",
          "stringValue" : "RedshiftCopyActivity",
          },
          {
          "key" : "insertMode",
          "stringValue" : "KEEP_EXISTING",
          },
          {
          "key" : "schedule",
          "refValue" : "DefaultScheduleTime",
          },
      ]
  },]

所以根据 RedShitCopyActivity 的文档，我们需要使用insertMode 来描述数据在复制到数据库表时的行为方式（插入/更新/删除），如下所示

insertMode : Determines what AWS Data Pipeline does with pre-existing data in the target table that overlaps with rows in the data to be loaded. Valid values are KEEP_EXISTING, OVERWRITE_EXISTING, TRUNCATE and APPEND. KEEP_EXISTING adds new rows to the table, while leaving any existing rows unmodified. KEEP_EXISTING and OVERWRITE_EXISTING use the primary key, sort, and distribution keys to identify which incoming rows to match with existing rows, according to the information provided in Updating and inserting new data in the Amazon Redshift Database Developer Guide. TRUNCATE deletes all the data in the destination table before writing the new data. APPEND will add all records to the end of the Redshift table. APPEND does not require a primary, distribution key, or sort key so items that may be potential duplicates may be appended.

那么我的要求是什么

1. 从 postgres（实际数据现在在 s3 中）复制到 Redshift 数据库时，如果发现已经存在的行，则只需更新它
2. 如果从 s3 中找到新记录，则在 Redshift 中创建新记录

但对我来说，即使我使用过KEEP_EXISTING 或OVERWRITE_EXISTING，数据只是一遍又一遍地重复，如上面的红移数据库图片所示

那么最后如何达到我的要求呢？是否还有任何调整或设置要添加到我的配置中？

编辑

redshift 中的表（公司）定义

Answer 1

如果要避免重复，则必须在 redshift 中定义 Primary key 并将 myInsertMode 设置为“OVERWRITE_EXISTING”。