ItemReader （地图）

分布式 Map 状态可以接受从工作流的上一个步骤中传递的 JSON 输入。

输入可以是 JSON 数组、JSON 对象或 JSON 对象节点内的数组。

Step Functions 将直接迭代数组的元素或 JSON 对象的键值对。

要从 JSON 对象中选择包含数组的特定节点，可以在 JSONata 状态Items字段中使用ItemsPath （ JSONPath 仅限地图）或使用 JSONata 表达式。

要处理单个项目，分布式地图状态会启动每个项目的子工作流程执行。以下选项卡显示了传递给 Map 状态的输入以及子工作流执行的相应输入的示例。

分布式 Map 状态 可以迭代存储在 Amazon S3 存储桶中的对象。当工作流程执行达到Map状态时，Step Functions 会调用 ListObjectsV2 API 操作，该操作会返回 Amazon S3 对象元数据的数组。在此数组中，每个项目都包含存储在存储桶中的实际数据的数据，例如ETag和密钥。

要处理数组中的各个项目，分布式 Map 状态 会启动一个子工作流执行。例如，假设您的 Amazon S3 存储桶包含 100 张图片。然后，调用 ListObjectsV2 API 操作后返回的数组包含 100 个元数据项。然后，分布式地图状态将启动 100 个子工作流程执行来处理每个项目。

要在没有嵌套工作流的情况下直接处理数据对象，可以选择 LOAD_AND_FLATTEN 转换选项来直接处理项目。

以下选项卡显示了该数据集的 ItemReader 字段语法和传递给子工作流执行的输入的示例。

通过增强对 S3 ListObjects V2 作为分布式地图中的输入源的支持，您的状态机可以直接读取和处理 Amazon S3 存储桶中的多个数据对象，无需使用嵌套地图来处理元数据！

使用该LOAD_AND_FLATTEN选项，您的状态机将执行以下操作：

使用转换选项，您不再需要嵌套分布式地图来处理元数据。使用 LOAD_AND_FLATTEN 选项可以提高可扩展性，减少活动地图的运行次数，并同时处理多个对象。

以下配置显示了的设置ItemReader：

"ItemReader": {
   "Resource": "arn:aws:states:::s3:listObjectsV2",
   "ReaderConfig": {
      "InputType": "JSON",
      "Transformation": "LOAD_AND_FLATTEN"
   },
   "Arguments": {
      "Bucket": "S3_BUCKET_NAME",
      "Prefix": "S3_BUCKET_PREFIX"
   }
}

分布式 Map 状态 可以接受存储在 Amazon S3 存储桶中的 JSON 文件作为数据集。JSON 文件必须包含数组或 JSON 对象。

当工作流程执行达到Map状态时，Step Functions 会调用 GetObjectAPI 操作来获取指定的 JSON 文件。

如果 JSON 文件包含嵌套对象结构，则可以选择带有数据集的特定节点ItemsPointer。例如，以下配置将提取库存中特色产品的嵌套列表。

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "JSON",
      "ItemsPointer": "/inventory/products/featured"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "nested-data-file.json"
   }
}

然后，Map 状态会迭代数组中的每个项目，并开始对每个项目执行子工作流。例如，如果您的 JSON 文件包含 1000 个数组项，则 Map 状态将启动 1000 个子工作流执行。

以下选项卡显示了该数据集的 ItemReader 字段语法和传递给子工作流执行的输入的示例。

在此示例中，假设您有一个名为 factcheck.json 的 JSON 文件。您已将此文件存储在 Amazon S3 存储桶中的名为 jsonDataset 的前缀中。以下是 JSON 数据集的示例：

[
  {
    "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"
  },
  ...
]

分布式地图状态可以接受存储在 Amazon S3 存储桶中的 JSON 行文件作为数据集。

以下选项卡显示了该数据集的 ItemReader 字段语法和传递给子工作流执行的输入的示例。

在本示例中，假设您有一个名为的 JSON 行文件factcheck.jsonl。您已将此文件存储在 Amazon S3 存储桶中的名为 jsonlDataset 的前缀中。以下是文件内容的示例。

{"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"}

分布式地图状态可以接受以文本分隔的文件作为数据集存储在 Amazon S3 存储桶中。如果您使用文本分隔文件作为数据集，则需要指定列标题。有关如何指定标题的信息，请参见该 ItemReader 字段的内容。

Step Functions 根据以下规则解析以文本分隔的文件：

有关 Step Functions 如何解析文本分隔文件的更多信息，请参阅。Example of parsing an input CSV file

当工作流程执行达到Map状态时，Step Functions 会调GetObject用 API 操作来获取指定的文件。然后，该Map状态会遍历文件中的每一行，并启动子工作流程执行以处理每行中的项目。例如，假设您提供了一个包含 100 行的文本分隔文件作为输入。然后，解释器将每一行传递给 Map 状态。Map 状态从标题行之后开始，按顺序处理项目。

以下选项卡显示了该数据集的 ItemReader 字段语法和传递给子工作流执行的输入的示例。

Parquet 文件可用作输入源。存储在 Amazon S3 中的 Apache Parquet 文件可大规模提供高效的列式数据处理。

使用 Parquet 文件时，以下条件适用：

以下显示了InputType设置为 Parquet 的 ASL 配置示例：

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "PARQUET"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "my-parquet-data-file-1.parquet"
   }
}

您可以使用根据查询结果生成UNLOAD的 Athena 清单文件来指定地图状态的数据文件来源。您可以ManifestType将设置为ATHENA_DATA、和InputTypeCSVJSONL、或Parquet。

运行UNLOAD查询时，除了实际的数据对象之外，Athena 还会生成一个数据清单文件。清单文件提供了数据文件的结构化 CSV 列表。清单和数据文件都保存到您在 Amazon S3 中的 Athena 查询结果位置。

UNLOAD (<YOUR_SELECT_QUERY>) TO 'S3_URI_FOR_STORING_DATA_OBJECT' WITH (format = 'JSON')

流程的概念概述，简而言之：

该功能可以处理 Athena 的 CSV、JSONL 和 Parquet 输出格式。单个清单文件中引用的所有对象的 InputType 格式必须相同。请注意，UNLOAD查询导出的 CSV 对象在第一行中不包含标题。看看CSVHeaderLocation是否需要提供列标题。

地图上下文还将包括，$states.context.Map.Item.Source因此您可以根据数据源自定义处理。

以下是配置为使用 Athena 清单的ItemReader配置示例：

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "ManifestType": "ATHENA_DATA",
      "InputType": "CSV | JSONL | PARQUET"
   },
   "Arguments": {
      "Bucket": "<S3_BUCKET_NAME>",
      "Key": "<S3_KEY_PREFIX><QUERY_ID>-manifest.csv"
   }
}

分布式 Map 状态 可以接受存储在 Amazon S3 存储桶中的 Amazon S3 清单文件作为数据集。

当工作流程执行达到Map状态时，Step Functions 会调用 GetObjectAPI 操作来获取指定的 Amazon S3 清单清单文件。

默认情况下，该Map状态随后会遍历清单中的对象，以返回一组 Amazon S3 清单对象元数据。

如果您指定 ManifestType 为 S3_INVENTORY，则 InputType 无法指定。

以下是 CSV 格式的清单文件示例：此文件包含名为 csvDataset 和 imageDataset 的对象，其中存储在名为 amzn-s3-demo-source-bucket 的 Amazon S3 存储桶中。

"amzn-s3-demo-source-bucket","csvDataset/","0","2022-11-16T00:27:19.000Z"
"amzn-s3-demo-source-bucket","csvDataset/titles.csv","3399671","2022-11-16T00:29:32.000Z"
"amzn-s3-demo-source-bucket","imageDataset/","0","2022-11-15T20:00:44.000Z"
"amzn-s3-demo-source-bucket","imageDataset/n02085620_10074.jpg","27034","2022-11-15T20:02:16.000Z"
...

以下 Amazon S3 库存清单文件示例显示了清单对象元数据的 CSV 标题。

{
  "sourceBucket" : "amzn-s3-demo-source-bucket",
  "destinationBucket" : "arn:aws:s3:::amzn-s3-demo-inventory",
  "version" : "2016-11-30",
  "creationTimestamp" : "1668560400000",
  "fileFormat" : "CSV",
  "fileSchema" : "Bucket, Key, Size, LastModifiedDate",
  "files" : [ {
    "key" : "amzn-s3-demo-bucket/destination-prefix/data/20e55de8-9c21-45d4-99b9-46c732000228.csv.gz",
    "size" : 7300,
    "MD5checksum" : "a7ff4a1d4164c3cd55851055ec8f6b20"
  } ]
}

以下选项卡显示了该数据集的 ItemReader 字段语法和传递给子工作流执行的输入的示例。