使用 AWS Step Functions 和 AWS Glue 编排基于 Amazon Redshift 的 ETL 工作流（三）

Step Functions 编排

现在您已有机会手动执行作业，可以转到由 Step Functions 编排的更具编程性的内容。

启动模板

我还提供了第三个 AWS CloudFormation 模板来启动此过程。它可以创建一个 Step Functions 状态机，调用刚创建的 AWS Glue Python Shell 作业的两个实例，以完成我在本博文开头概述的两项任务。

对于 BucketName，粘贴在第二个 AWS CloudFormation 堆栈中创建的脚本存储桶的名称。对于 GlueJobName，输入刚创建作业的名称。其他信息保留为默认值，如以下屏幕截图所示。启动堆栈并等待其显示创建完成（这只需几分钟），然后再转到下一部分。

使用 Step Functions 状态机

状态机由一系列步骤组成，使您可以将服务拼接到稳健的 ETL 工作流中。您可以在执行过程中监控每个步骤，这意味着您可以在 ETL 工作流中快速发现并解决问题，最终自动完成。

看一下刚启动的状态机以更深入地了解。导航到 AWS 控制台中的 Step Functions 并查找名称类似 GlueJobStateMachine-###### 的状态机。 选择编辑以查看状态机配置，如以下屏幕截图所示。

它的外观应与以下屏幕截图相同：

如您所见，状态机是使用由任务定义和工作流逻辑组成的 JSON 模板创建的。您可以运行并行任务，发现错误，甚至暂停工作流并等待手动回调继续。我提供的示例包含两个用于运行 SQL 语句的任务，这些任务可以完成我在博文开头介绍的目标：

1. 使用 Redshift Spectrum 从 S3 加载数据

2. 转换数据并将其写回到 S3

每个任务都包含基本的错误处理，如果发现这些错误，则会将工作流路由到错误通知任务。本示例是一个展示如何构建基本工作流简单示例，但是您可以参考 Step Functions 文档了解更复杂的工作流示例，以帮助构建稳健的 ETL 管道。Step Functions 还支持通过嵌套工作流重复使用模块化组件。

SQL 审查

状态机将检索并运行以下 SQL 语句：

SQL

INSERT INTO reviewsSELECT marketplace, customer_id, review_id, product_id, product_parent, product_title, star_rating, helpful_votes, total_votes, vine, verified_purchase, review_date, year, product_categoryFROM amzreviews.reviewsWHERE year > 2015;

如前所述，Amazon Redshift Spectrum 是使用 INSERT INTO 语句运行 ETL 的绝佳方法。该示例是对数据的简单加载，如图在 S3 中一样，但是请记住，可以在加载之前添加更复杂的 SQL 语句来转换数据。

SQL

UNLOAD ('SELECT marketplace, product_category, product_title, review_id, helpful_votes, AVG(star_rating) as average_stars FROM reviews GROUP BY marketplace, product_category, product_title, review_id, helpful_votes ORDER BY helpful_votes DESC, average_stars DESC')TO 's3://bucket/testunload/'iam_role 'rolearn';

该语句按产品分类评论，按有用投票数排序，并使用 UNLOAD 写入 Amazon S3。

状态机执行

现在一切就绪，开始执行。从状态机主页选择开始执行。

保持默认值不变，然后选择开始以开始执行。执行开始后，您将进入可视化工作流界面，在其中可以跟踪执行进度，如以下屏幕截图所示。

每个查询都需要几分钟才能运行。同时，您可以观看 Amazon Redshift 查询日志，以实时跟踪查询进度。可通过在 AWS 控制台中导航到 Amazon Redshift，选择您的 Amazon Redshift 集群，然后选择查询选项卡来找到查询日志，如以下屏幕截图所示。

在两次查询都看到 COMPLETED 之后，返回状态机执行。您应该看到每种状态已成功，如以下屏幕截图所示。

接下来，导航到 S3 AWS 控制台页面中的数据存储桶（请参阅 CloudFormation 资源选项卡中的 DataBucket）。如果一切按计划进行，您将在存储桶中看到一个名为 testunload 的文件夹，其中包含卸载数据，如以下屏幕截图所示。

将故障注入 Step Functions 状态机

接下来，通过有意引起错误来测试状态机的错误处理组件。一种简单的方法是编辑状态机，并在 ReadFilterJob 任务中错误拼写 Secrets Manager 密钥的名称，如以下屏幕截图所示。

如果您希望将错误输出发送给您，可选择订阅错误通知 SNS 主题。按照之前的操作，开始另一个状态机执行。这次，工作流应选择通向 NotifyFailure 任务的路径，如以下屏幕截图所示。如果您订阅了与其关联的 SNS 主题，此后不久应该会收到一条消息。

状态机日志将更详细地显示错误，如以下屏幕截图所示。

小结

在本博文中，我演示了如何使用无服务器的 AWS Step Functions 和 AWS Glue Python Shells 作业来编排基于 Amazon Redshift 的 ETL。正如我在简介中提到的那样，这些概念也可以更普遍地应用于其他基于 SQL 的 ETL，因此，今天就可以使用它们开始构建您自己的基于 SQL 的 ETL 管道！

作者介绍：

**

Ben Romano 是 AWS 的数据实验室解决方案架构师。Ben 在 AWS 数据实验室中仅用四天的时间就可以帮助我们的客户设计和构建数据及分析原型。

**

本文转载自 AWS 技术博客。

原文链接：https://amazonaws-china.com/cn/blogs/china/orchestrate-amazon-redshift-based-etl-workflows-with-aws-step-functions-and-aws-glue/