在 Step Functions 中使用分布式 Map 复制大规模 CSV 数据

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

2. 在 选择模板对话框中，选择空白。

3. 选择选择，以便在设计模式下打开工作流程工作室。

4. 从流选项卡中，将 Map 状态拖放到标有将第一个状态拖至此处的空白状态处。

5. 在配置选项卡下，在状态名称中输入 Process data。

6. 从操作选项卡中，将 Amazon Lambda 调用 API 操作拖放到 Process data 状态中。

7. 将 Amazon Lambda 调用状态重命名为 Process CSV data。

要配置必填句字段，请执行以下操作：

1. 选择 Process data 状态，然后在配置选项卡中执行以下操作：

   1. 对于处理模式，选择分布式。

   2. 对于项目来源，选择 Amazon S3，然后从 S3 项目来源下拉列表中选择 S3 中的 CSV 文件。

   3. 执行以下操作来指定 CSV 文件的 Amazon S3 位置：

      1. 对于 S3 对象，请从下拉列表中选择输入存储桶和密钥。

      2. 对于存储桶，输入包含 CSV 文件的 Amazon S3 存储桶的名称。例如，amzn-s3-demo-source-bucket。

      3. 对于密钥，输入保存 CSV 文件的 Amazon S3 对象的名称。您还必须在此字段中指定 CSV 文件的名称。例如，csvDataset/ratings.csv。

   4. 对于 CSV 文件，还必须指定列标题的位置。为此，请选择其他配置，如果 CSV 文件的第一行是标题，对于 CSV 标题位置，请保留第一行的默认选择。否则，请选择给定以在状态机定义中指定标题。有关更多信息，请参阅 ReaderConfig。

   5. 对于子执行类型，请选择快速。

2. 在导出位置中，要将 Map Run 结果导出到特定的 Amazon S3 位置，请选择将 Map 状态的输出导出到 Amazon S3。

3. 执行以下操作：

   1. 对于 S3 存储桶，请从下拉列表中选择输入存储桶名称和前缀。

   2. 对于存储桶，输入要将结果导出到的 Amazon S3 存储桶的名称。例如，mapOutputs。

   3. 对于前缀，输入要将结果保存到的文件夹名称。例如，resultData。

1. 选择 Process data 状态。然后，在项目来源中，选择其他配置。

2. 执行以下操作：

   1. 选择使用 ItemSelector 修改项目，为每个子工作流执行指定自定义 JSON 输入。

   2. 输入以下 JSON 输入：

      {
        "index.$": "$$.Map.Item.Index",
        "value.$": "$$.Map.Item.Value"
      }

      有关如何创建自定义输入的信息，请参阅 ItemSelector （地图）。

3. 在运行时设置中，对于并发限制，指定分布式 Map 状态 可以启动的并发子工作流执行数量。例如，输入 100。

4. 在浏览器中打开一个新窗口或选项卡，完成您将在此工作流中使用的 Lambda 函数的配置，如第 4 步：配置 Lambda 函数中所述。

1. 打开 Lambda 控制台，然后选择创建函数。

2. 在创建函数页面上，选择从头开始创作。

3. 在基本信息部分中，配置您的 Lambda 函数：

   1. 对于函数名称，请输入 distributedMapLambda。

   2. 对于 Runtime (运行时)，选择 Node.js。

   3. 保留所有默认选项，然后选择创建函数。

   4. 创建 Lambda 函数后，复制页面右上角显示的函数 Amazon 资源名称 (ARN)。您需要在工作流原型中提供此信息。以下是示例 ARN：

      arn:aws:lambda:us-east-2:123456789012:function:distributedMapLambda

4. 复制以下 Lambda 函数代码，并将其粘贴到 distributedMapLambda 页面的代码源部分。

   exports.handler = async function(event, context) {
       console.log("Received Input:\n", event);
   
       return {
           'statusCode' : 200,
           'inputReceived' : event //returns the input that it received
       }
   };

5. 选择部署。函数部署后，选择测试，查看您的 Lambda 函数的输出。

1. 返回到创建工作流原型的选项卡或窗口。

2. 选择处理 CSV 数据步骤，然后在配置选项卡中执行以下操作：

   1. 对于集成类型，请选择已优化。

   2. 对于函数名称，输入 Lambda 函数名称。从出现的下拉列表中选择函数，或者选择输入函数名称并提供 Lambda 函数 ARN。

1. （可选）在 检查器面板 面板上选择定义，然后查看状态机定义。

   提示

   您也可以在 Workflow Studio 的代码编辑器中查看 ASL 的定义。在代码编辑器中，还可以编辑工作流的 ASL 定义。

   以下示例代码显示了为您的工作流自动生成的 Amazon States Language 定义。

   {
     "Comment": "Using Map state in Distributed mode",
     "StartAt": "Process data",
     "States": {
       "Process data": {
         "Type": "Map",
         "MaxConcurrency": 100,
         "ItemReader": {
           "ReaderConfig": {
             "InputType": "CSV",
             "CSVHeaderLocation": "FIRST_ROW"
           },
           "Resource": "arn:aws:states:::s3:getObject",
           "Parameters": {
             "Bucket": "amzn-s3-demo-source-bucket",
             "Key": "csvDataset/ratings.csv"
           }
         },
         "ItemProcessor": {
           "ProcessorConfig": {
             "Mode": "DISTRIBUTED",
             "ExecutionType": "EXPRESS"
           },
           "StartAt": "Process CSV data",
           "States": {
             "Process CSV data": {
               "Type": "Task",
               "Resource": "arn:aws:states:::lambda:invoke",
               "OutputPath": "$.Payload",
               "Parameters": {
                 "Payload.$": "$",
                 "FunctionName": "arn:aws:lambda:us-east-2:123456789012:function:distributedMapLambda"
               },
               "End": true
             }
           }
         },
         "Label": "Processdata",
         "End": true,
         "ResultWriter": {
           "Resource": "arn:aws:states:::s3:putObject",
           "Parameters": {
             "Bucket": "mapOutputs",
             "Prefix": "resultData"
           }
         },
         "ItemSelector": {
           "index.$": "$$.Map.Item.Index",
           "value.$": "$$.Map.Item.Value"
         }
       }
     }
   }

2. 为状态机指定一个名称。要执行此操作，请选择默认状态机名称 MyStateMachine 旁边的编辑图标。然后，找到状态机配置，在状态机名称框中指定一个名称。

   对于本教程，请输入名称 DistributedMapDemo。

3. （可选）在状态机配置中，指定其他工作流设置，例如状态机类型及其执行角色。

   在本教程中，请保留状态机配置中的所有默认选项。

4. 在确认角色创建对话框中，选择确认继续。

   您也可以选择查看角色设置，返回至状态机配置。

   注意

   如果您删除了 Step Functions 创建的 IAM 角色，Step Functions 在以后无法重新创建该角色。同样，如果您修改了该角色（例如，通过在 IAM 策略中从主体中删除 Step Functions），Step Functions 在以后也无法还原其原始设置。

1. 在 DistributedMapDemo 页面上，选择启动执行。

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

   1. （可选）输入自定义执行名称，以便覆盖生成的默认执行名称。

      非 ASCII 名称和日志记录

      Step Functions 对于状态机、执行、活动和标签接受包含非 ASCII 字符的名称。由于此类字符不适用于 Amazon CloudWatch，因此我们建议您仅使用 ASCII 字符，这样您就可以在 CloudWatch 中跟踪指标。

   2. （可选）在输入框中，以 JSON 格式输入输入值以便运行工作流。

   3. 选择启动执行。

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

      要查看执行结果，请在图表视图上选择各个状态，然后在步骤详细信息窗格中选择各个选项卡，分别查看每个状态的详细信息，包括输入、输出和定义。有关可在执行详细信息页面上查看的执行信息的详细信息，请参阅执行详细信息概览。

   例如，选择 Map 状态，然后选择 Map Run，打开 Map Run 详细信息页面。在此页面上，您可以查看分布式 Map 状态的所有执行细节及其启动的子工作流执行。有关该页面的信息，请参阅查看 Map Run。