AWS:通过amazone-data-pipeline将数据从S3传输到Redshift,而不是COPY的其他功能

Question

我正在尝试使用Amazon-Data-Pipeline工具将数据从Amazon S3-Cloud传输到Amazon-Redshift.

是否可以在传输数据时使用eG和SQL语句更改数据,以便只有SQL-Statement的结果才能成为Redshift的输入？

我只发现了Copy命令:

  {
    "id": "S3Input",
    "type": "S3DataNode",
    "schedule": {
    "ref": "MySchedule"
  },
  "filePath": "s3://example-bucket/source/inputfile.csv"
},

资料来源:http://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-get-started-copy-data-cli.html

Answer 1

对的,这是可能的.它有两种方法:

1. 使用transformSQL的RedShiftCopyActivity

transformSQL如果转换是在记录范围内进行的,并且是及时加载的,例如每天或每小时.这样,更改仅应用于批处理,而不应用于整个表.

以下是文档的摘录:

transformSql: 用于转换输入数据的SQL SELECT表达式.从DynamoDB或Amazon S3复制数据时,AWS Data Pipeline会创建一个名为staging的表,并最初将其加载到那里.此表中的数据用于更新目标表.如果指定了transformSql选项,则会从指定的SQL语句创建第二个临时表.然后,在最终目标表中更新来自第二个登台表的数据.因此,必须在名为staging的表上运行transformSql,并且transformSql的输出模式必须与最终目标表的模式匹配.

请在下面找到transformSql的使用示例.请注意,select来自staging表.它会有效地运行CREATE TEMPORARY TABLE staging2 AS SELECT <...> FROM staging;.此外,必须包含所有字段并匹配RedShift DB中的现有表.

{
  "id": "LoadUsersRedshiftCopyActivity",
  "name": "Load Users",
  "insertMode": "OVERWRITE_EXISTING",
  "transformSql": "SELECT u.id, u.email, u.first_name, u.last_name, u.admin, u.guest, CONVERT_TIMEZONE('US/Pacific', cs.created_at_pst) AS created_at_pst, CONVERT_TIMEZONE('US/Pacific', cs.updated_at_pst) AS updated_at_pst FROM staging u;",
  "type": "RedshiftCopyActivity",
  "runsOn": {
    "ref": "OregonEc2Resource"
  },
  "schedule": {
    "ref": "HourlySchedule"
  },
  "input": {
    "ref": "OregonUsersS3DataNode"
  },
  "output": {
    "ref": "OregonUsersDashboardRedshiftDatabase"
  },
  "onSuccess": {
    "ref": "LoadUsersSuccessSnsAlarm"
  },
  "onFail": {
    "ref": "LoadUsersFailureSnsAlarm"
  },
  "dependsOn": {
    "ref": "BewteenRegionsCopyActivity"
  }
}

2. 使用script的SqlActivity

SqlActivity允许对整个数据集进行操作,并且可以安排在特定事件之后通过dependsOn机制运行

{
  "name": "Add location ID",
  "id": "AddCardpoolLocationSqlActivity",
  "type": "SqlActivity",
  "script": "INSERT INTO locations (id) SELECT 100000 WHERE NOT EXISTS (SELECT * FROM locations WHERE id = 100000);",
  "database": {
    "ref": "DashboardRedshiftDatabase"
  },
  "schedule": {
    "ref": "HourlySchedule"
  },
  "output": {
    "ref": "LocationsDashboardRedshiftDatabase"
  },
  "runsOn": {
    "ref": "OregonEc2Resource"
  },
  "dependsOn": {
    "ref": "LoadLocationsRedshiftCopyActivity"
  }
}