使用分布式 Map 处理 Amazon S3 存储桶中的数据 - Amazon Step Functions
Amazon CloudFormation 模板和其他资源第 1 步：创建状态机并预置资源第 2 步：运行状态机
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

使用分布式 Map 处理 Amazon S3 存储桶中的数据

此示例项目演示了如何使用分布式 Map 状态 处理大规模数据，例如，分析历史天气数据并确定每月地球上平均温度最高的气象站。天气数据记录在 1.2 万多个 CSV 文件中，这些文件又存储在 Amazon S3 存储桶中。

此示例项目包括两个分布式 Map 状态，分别名为 Distributed S3 copy NOA Data 和 ProcessNOAAData。分布式 S3 副本 NOA 数据会迭代名noaa-gsod-pds为的 Amazon S3 公共存储桶中的 CSV 文件，并将它们复制到您的中的 Amazon S3 存储桶中。 Amazon Web Services 账户ProcessNOAAData 对复制的文件进行迭代，并包含一个用于执行温度分析的 Lambda 函数。

示例项目首先通过调用 ListObjectsV2 API 操作来检查 Amazon S3 存储桶的内容。根据响应此调用返回的密钥数量，示例项目会做出以下决定之一：

  • 如果密钥计数大于或等于 1，则项目将转换到 ProcessNOAAData 状态。此分布式地图状态包括一个名为的函数 TemperatureFunction，该Lambda函数用于查找每月平均温度最高的气象站。此函数返回一个以 year-month 为密钥的目录和一个包含将气象站相关信息作为值的目录。

  • 如果返回的密钥计数不超过 1，则分布式 S3 副本 NOA 数据状态将列出公共存储桶中的所有对象，noaa-gsod-pds并以 100 个为一批迭代地将各个对象复制到您账户中的另一个存储桶。内联 Map 会执行对象的迭代复制。

    复制完所有对象后，项目将转换到 ProcessNOAAData 状态，处理天气数据。

示例项目最终转换为 reduc Lambda er 函数，该函数对TemperatureFunction函数返回的结果进行最终聚合，并将结果写入Amazon DynamoDB表中。

使用分布式 Map，一次最多可运行 1 万个并行子工作流。在此示例项目中，ProcessNOAAData 分布式 Map 的最大并发量设置为 3000，这就限制了它只能并行执行 3000 个子工作流。

此示例项目创建状态机、支持 Amazon 资源并配置相关的 IAM 权限。探索此示例项目，了解如何使用分布式 Map 编排大规模并行工作负载，或将其作为您自己项目的起点。

重要

此示例项目仅在美国东部（弗吉尼亚州北部）区域可用。

Amazon CloudFormation 模板和其他资源

您可以使用 CloudFormation 模板来部署此示例项目。此模板在您的中创建了以下资源 Amazon Web Services 账户：

  • 一个 Step Functions 状态机。

  • 状态机的执行角色。此角色授予您的状态机访问其他 Amazon Web Services 资源所需的权限，例如 Lambda 函数的 Invoke 操作。

  • 一个名为 NOAADataBucket 的 Amazon S3 存储桶。此存储桶包含带有天气数据的 CSV 文件。

  • 一个名为 ReducerFunction 的 Lambda 函数，用于对天气数据进行最终聚合，并将结果写入 Amazon DynamoDB 表。

  • reducer Lambda 函数的执行角色。此角色授予函数访问其他角色的权限 Amazon Web Services。

  • 一个名为 ResultsBucket 的 Amazon S3 输出桶，用于存储天气分析结果。

  • 一个名为 ResultsDynamoDBTable 的 DynamoDB 表，其中包含 ReducerFunction 返回的结果。

  • 一个名为 TemperatureFunction 的 Lambda 函数，用于查找最高月平均气温。

  • Lambda 函数的执行角色。此角色授予函数访问其他角色的权限 Amazon Web Services。

  • 存储与状态机执行历史相关的信息的 CloudWatch 日志组。

重要

标准费用适用于每项服务。

第 1 步：创建状态机并预置资源

  1. 打开 Step Functions 控制台，然后选择创建状态机。

  2. 在搜索框中键入 Distributed Map to process files in S3，然后从返回的搜索结果中选择分布式 Map 处理 S3 中的文件。

  3. 选择下一步以继续。

  4. Step Functions 列出了您选择的示例项目中 Amazon Web Services 使用的。它还显示了示例项目的工作流图。将此项目部署到您的， Amazon Web Services 账户 或者将其用作构建您自己的项目的起点。根据您想继续的方式，选择运行演示或构建依据。

    有关将为此示例项目创建的资源信息，请参阅 Amazon CloudFormation 模板和其他资源。

    下图显示了分布式 Map 处理 S3 中的文件示例项目的工作流图：

    分布式 Map 处理 S3 中的文件示例项目的工作流图。

  5. 选择使用模板继续进行选择。

  6. 请执行以下操作之一：

    • 如果您选择构建依据，Step Functions 将为您选择的示例项目创建工作流原型。Step Functions 不会部署工作流定义中列出的资源。

      在 Workflow Studio 的设计模式下，从状态浏览器中拖放状态，继续构建工作流原型。或者切换到代码模式，该模式提供了一个类似于 VS Code 的集成代码编辑器，用于在 Step Functions 控制台中更新状态机的 Amazon States Language (ASL) 定义。有关使用 Workflow Studio 构建状态机的更多信息，请参阅使用 Workflow Studio。

      重要

      请记住，在运行工作流之前，为示例项目中使用的资源更新占位符 Amazon 资源名称 (ARN)。

    • 如果您选择了 “运行演示”，Step Functions 将创建一个只读示例项目，该项目使用 Amazon CloudFormation 模板将该模板中列出的 Amazon 资源部署到您的 Amazon Web Services 账户。

      提示

      要查看示例项目的状态机定义，请选择代码。

      准备就绪后，选择部署并运行以部署示例项目并创建资源。

      创建这些资源和相关 IAM 权限可能需要长达 10 分钟的时间。在部署资源时，您可以打开 CloudFormation 堆栈 ID 链接以查看正在配置哪些资源。

      创建示例项目中的所有资源后，您可以在状态机页面上看到新的示例项目。

      重要

      CloudFormation 模板中使用的每项服务都可能收取标准费用。

第 2 步：运行状态机

配置并部署完所有资源后，您可以运行状态机。

  1. 在状态机页面上，选择您的示例项目。

  2. 在示例项目页面上，选择启动执行。

  3. 在启动执行对话框中，执行以下操作：

    1. （可选）以 JSON 格式键入输入值以运行您的示例项目。

      如果您选择运行演示，则无需提供任何执行输入。

      注意

      如果您部署的演示项目包含预先填充的执行输入数据，请使用该输入来运行状态机。

    2. 选择启动执行。

    3. （可选）Step Functions 控制台会将您引导到一个以您的执行 ID 为标题的页面。该页面被称为执行详细信息页面。在此页面上，您可以随着执行的进展或者在执行完成后查看执行结果。

      执行完成后，请在图表视图上选择各个状态，然后在步骤详细信息窗格中选择各个选项卡，分别查看每个状态的详细信息，包括输入、输出和定义。

    4. （可选）查看导出到 Amazon S3 存储桶的执行结果。这些结果包括数据，例如执行输入和输出、ARN 和执行状态。有关更多信息，请参阅 ResultWriter。