bash - AWS Data Pipeline - 如何从 ShellCommandActivity 设置全局管道变量

Question

我正在尝试扩充我的管道(将数据从 RDS 迁移到 RedShift)，以便它选择 id 大于 RedShift 中存在的最大 id 的所有行.我在 Python 中有一个脚本可以计算这个值并将其返回到输出。我想获取此输出并将其保存到一个变量 max_id 中，稍后我可以在我的 RDS 选择查询中引用该变量。例如，我的 RDS 选择部分目前如下所示:

{
  "database": {
    "ref": "rds_mysql"
  },
  "scheduleType": "TIMESERIES",
  "name": "SrcRDSTable",
  "id": "SrcRDSTable",
  "type": "SqlDataNode",
  "table": "#{myRDSTableName}",
  "selectQuery": "select * from #{table} where #{myRDSTableLastModifiedCol} > '#{max_id}'"
},

然后我想在此之前添加一个部分来执行 bash 脚本，检索 id 字段并将其保存到变量 max_id 以便可以引用它在上面的代码中。到目前为止，我有:

{
 "myComment": "Retrieves the maximum ID for a given table in RedShift",
  "id": "ShellCommandActivity_Max_ID",
  "workerGroup": "wg-12345",
  "type": "ShellCommandActivity",
  "command": "starting_point=$(/usr/bin/python /home/user/aws-taskrunner-docker/get_id.py --schema=schema_name --table=users --database=master)"
},

我如何调整以上设置max_id为starting_point的值？谢谢。

Answer 1

不幸的是，我认为没有办法在管道执行期间设置管道参数。以下是可能对您有所帮助的几个选项。

首先，如果你的数据表有一列修改日期，那么你可以使用管道模板Incremental Copy of RDS MySQL to Redshift .如果您不使用 MySQL，您仍然可以根据需要修改该模板。

或者，您可以创建一个使用 python 连接到您的 RDS 数据库并将相关记录集导出到 S3 的 ShellCommandActivity，而不是使用 SqlDataNode。然后，您可以使用 RedshiftCopyActivity 从 S3 导入记录。