在分布式模式下使用地图状态

使用该ItemReader字段指定数据集的位置，该Map州从中读取其输入数据。以下示例显示如何通过指定存储 CSV 文件的 Amazon S3 存储桶名称和对象密钥来将 CSV 文件用作数据集。

{
  "Map": {
    "Type": "Map",
    ...
    "ItemReader": {
      "ReaderConfig": {
        "InputType": "CSV",
        "CSVHeaderLocation": "FIRST_ROW"
      },
      "Resource": "arn:aws:states:::s3:getObject",
      "Parameters": {
        "Bucket": "Database2022",
        "Key": "csv-dataset/ratings.csv"
      }
    },
    ...
    "End": true
  }
}

使用该MaxItems字段的子ItemReader字段限制该Map州可以从数据集中读取的项目数量。例如，如果您将 CSV 数据集的MaxItems子字段值指定为 90，则该Map州仅读取 CSV 文件的前 90 行，从标题行之后开始。

{
  "Map": {
    "Type": "Map",
    ...
    "ItemReader": {
      "ReaderConfig": {
        "MaxItems": 90,
        "InputType": "CSV",
        "CSVHeaderLocation": "FIRST_ROW"
      },
      ...
    },
    ...
    "End": true
  }
}

如果您的数据集是从工作流程中先前步骤传递的 JSON 输入，并且它包含一个数组，请使用该ItemsPath字段选择包含该数组的节点。

例如，给定以下 JSON 输入：

{
  "facts": [
    {
      "verdict": "true",
      "statement_date": "6/11/2008",
      "statement_source": "speech"
    },
    {
      "verdict": "false",
      "statement_date": "6/7/2022",
      "statement_source": "television"
    },
    {
      "verdict": "mostly-true",
      "statement_date": "5/18/2016",
      "statement_source": "news"
    }
  ]
}

使用以下示例所示的ItemsPath字段选择包含数组的 JSON 节点：

"Map": {
  "Type": "Map",
  "ItemProcessor": {
    "ProcessorConfig": {
      "Mode": "DISTRIBUTED",
      ...
    }
  },
  ...
  "ItemsPath": "$.facts",
  ...
  "End": true
}

在将单个数据集项目的值传递给Map状态迭代之前，使用该ItemSelector字段覆盖这些项目的值。要覆盖这些值，您可以指定包含键值对集合。这些对可以是您在状态机定义中提供的静态值，也可以是使用路径从状态输入中选择的值，也可以是从上下文对象访问的值。

例如，以下自定义 JSON 输入在传递到Map状态迭代之前替换了原始输入。每个子工作流程执行都会收到以下包含静态值和两个上下文对象数据项的自定义输入。

{
  "Map": {
    "Type": "Map",
    ...
    "ItemSelector": {
      "foo": "bear",
      "index.$": "$$.Map.Item.Index",
      "value.$": "$$.Map.Item.Value"
    },
    ...
    "End": true
  }
}

默认情况下，该Map状态将数据集中的每个项目作为输入传递给单个子工作流程执行。使用该ItemBatcher字段处理每个子工作流程执行中的一组项目。在此字段中，您可以指定要批处理的最大项目数、以字节为单位的最大批处理大小，或者这两个值。解释器将指定数量的项目添加到 Items 数组中。然后，它将数组作为输入传递给每个子工作流程执行。例如，如果您将要批处理的最大项目数指定为 10，则解释器会将指定数量的项目添加到每个子工作流程执行的输入中的 Items 数组中。

以下示例说明如何通过指定MaxItemsPerBatch字段来批处理 10 个数据集项目：

{
  "Map": {
    "Type": "Map",
    "ItemBatcher": {
      "MaxItemsPerBatch": 10
    },
    ...
    "End": true
  }
}

以下示例显示了Items数组内的一批项目，Map状态将其作为输入传递给子工作流程执行：

{
  "Items": [
    {
      "rating": "3.0",
      "movieId": "1244",
      "userId": "2",
      "timestamp": "1192913551"
    },
    {
      "rating": "4.5",
      "movieId": "1296",
      "userId": "2",
      "timestamp": "1192913608"
    },
    ...
  ]
}

使用ResultWriter字段将所有子工作流程执行的结果导出Amazon S3 存储桶。如果您的输出有效负载大小超过 256 KB，将结果导出Amazon S3 存储桶会很有帮助。要导出所有子工作流程执行的结果，请指定要将结果存储到的 Amazon S3 存储桶名称和对象前缀。

{
  "Map": {
    "Type": "Map",
    ...
    "ResultWriter": {
      "Resource": "arn:aws:states:::s3:putObject",
      "Parameters": {
        "Bucket": "processedOutput",
        "Prefix": "test-run"
      }
    },
    ...
    "End": true
  }
}