亚马逊数据流水线:通过使用amazon- data - S3将数据从数据传输到红移的复制以外的其他功能

Question

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

在传输数据时，是否可以使用SQL语句来更改数据，以便只将SQL语句的结果作为Redshift的输入？

我只找到了这样的复制命令：

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

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

Answer 1

是的，这是可能的。有两种方法可以解决这个问题：

使用RedShiftCopyActivity的transformSQL的

如果转换是在记录的范围内执行的，那么transformSQL是很有用的，这些记录会及时加载，例如每天或每小时。这样，更改只会应用于批次，而不会应用于整个表。

以下是文档的摘录：

transformSql:用于转换输入数据的SQL SELECT表达式。当您从DynamoDB或亚马逊S3复制数据时，亚马逊网络服务数据管道将创建一个名为staging的表，并最初将其加载到其中。此表中的数据用于更新目标表。如果指定了SQL选项，则从指定的transformSql语句创建第二个临时表。然后，在最终目标表中更新来自第二个临时表的数据。因此，必须在名为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"
  }
}

使用SqlActivity的script的

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"
  }
}

Answer 2

在RedshiftCopyActivity中有一个可选的字段，叫做'transformSql‘。

http://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-redshiftcopyactivity.html

我个人并没有使用过它，但是从外观上看，它看起来像是-您将把您的s3数据放在一个临时表中，并且这个sql stmt将返回转换后的数据以供redshift插入。

因此，您需要在选择中列出所有字段，无论您是否要转换该字段。

Answer 3

AWS Datapipeline SqlActivity

{
  "id" : "MySqlActivity",
  "type" : "SqlActivity",
  "database" : { "ref": "MyDatabase" },
  "script" : "insert into AnalyticsTable (select (cast(requestEndTime as bigint) - cast(requestBeginTime as bigint)) as requestTime, hostname from StructuredLogs where hostname LIKE '%.domain.sfx');",
  "schedule" : { "ref": "Hour" },
  "queue" : "priority"
}

因此，基本上在SQL“脚本”中，任何脚本/转换/命令Amazon Redshift SQL Commands

transformSql很好，但只支持用于转换输入数据的SQL SELECT表达式。参考：RedshiftCopyActivity