Amazon Redshift - 卸载到 S3 - 动态 S3 文件名

Posted 2023-03-31

【中文标题】Amazon Redshift - 卸载到 S3 - 动态 S3 文件名

【英文标题】：Amazon Redshift - Unload to S3 - Dynamic S3 file name

【发布时间】：2016-04-14 07:47:37

【问题描述】：

我已经在Redshift中使用UNLOAD语句有一段时间了，它可以更容易地将文件转储到S3然后让人们进行分析。

是时候尝试自动化它了。我们让Amazon Data Pipeline 运行多个任务，我想运行SQLActivity 来自动执行UNLOAD。我使用托管在S3 中的SQL 脚本。

查询本身是正确的，但我一直想弄清楚如何动态分配文件名。例如：

UNLOAD('<the_query>')
TO 's3://my-bucket/' || to_char(current_date)
WITH CREDENTIALS '<credentials>'
ALLOWOVERWRITE
PARALLEL OFF

不起作用，当然我怀疑您无法在“TO”行中执行函数（to_char）。有没有其他方法可以做到？

如果UNLOAD 不可行，我还有其他选择如何使用当前可用的基础架构自动执行此类任务（Redshift + S3 + Data Pipeline，我们的Amazon EMR 尚未激活） .

我认为唯一可行（但不确定）不是使用脚本，而是将脚本复制到SQLActivity 中的Script 选项中（此时它指向一个文件）并引用@ 987654338@

【参考方案1】：

为什么不使用 RedshiftCopyActivity 从 Redshift 复制到 S3？输入是 RedshiftDataNode，输出是 S3DataNode，您可以在其中指定 directoryPath 的表达式。

还可以在RedshiftCopyActivity中指定transformSql属性覆盖默认值：select * from + inputRedshiftTable。

示例管道：

“对象”：[ "id": "CSVId1", “名称”：“默认CSV1”， “类型”：“CSV” , "id": "RedshiftDatabaseId1", “数据库名称”：“数据库名称”， “用户名”：“用户”， "name": "DefaultRedshiftDatabase1", "*密码": "密码", "type": "RedshiftDatabase", “clusterId”：“redshiftclusterId” , "id": "默认", "scheduleType": "时间序列", "failureAndRerunMode": "CASCADE", “名称”：“默认”， "role": "DataPipelineDefaultRole", "resourceRole": "DataPipelineDefaultResourceRole" , "id": "RedshiftDataNodeId1", “日程”： "ref": "ScheduleId1" , "tableName": "订单", "name": "DefaultRedshiftDataNode1", "type": "RedshiftDataNode", “数据库”： "ref": "RedshiftDatabaseId1" , "id": "Ec2ResourceId1", “日程”： "ref": "ScheduleId1" , "securityGroups": "MySecurityGroup", "name": "DefaultEc2Resource1", "role": "DataPipelineDefaultRole", "logUri": "s3://myLogs", "resourceRole": "DataPipelineDefaultResourceRole", “类型”：“Ec2Resource” , "myComment": "该对象用于控制任务调度。", "id": "DefaultSchedule1", "name": "RunOnce", “发生”：“1”， "期间": "1 天", “类型”：“时间表”， "startAt": "FIRST_ACTIVATION_DATE_TIME" , "id": "S3DataNodeId1", “日程”： "ref": "ScheduleId1" , "directoryPath": "s3://my-bucket/#format(@scheduledStartTime, 'YYYY-MM-dd-HH-mm-ss')", "name": "DefaultS3DataNode1", “数据格式”： “参考”：“CSVId1” , “类型”：“S3DataNode” , "id": "RedshiftCopyActivityId1", “输出”： “参考”：“S3DataNodeId1” , “输入”： "ref": "RedshiftDataNodeId1" , “日程”： "ref": "ScheduleId1" , "name": "DefaultRedshiftCopyActivity1", “运行”： “参考”：“Ec2ResourceId1” , “类型”：“RedshiftCopyActivity” ]

【讨论】：

如何根据列值卸载分区中的数据？使用文件路径中的列名，比如写入按事务日期或类似分区的文件路径？

【参考方案2】：

您可以通过 SSH 连接到集群吗？如果是这样，我建议编写一个 shell 脚本，您可以在其中创建变量等，然后将这些变量传递到连接的语句查询中

【参考方案3】：

通过在 unload 语句周围使用 redshift 过程包装器并动态派生 s3 路径名。

执行动态查询并在您的作业中调用动态创建 UNLOAD 语句并执行该语句的过程。

这样您就可以避开其他服务。但取决于您正在处理什么样的用例。

【讨论】：

您有文档链接吗？