ItemReader（Map）

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

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

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

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

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

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

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

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

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

在分布式 Map 中，对 S3 ListObjectsV2 作为输入源的支持得到了增强，使您的状态机可以直接读取和处理 Amazon S3 存储桶中的多个数据对象，而无需使用嵌套映射来处理元数据！

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

使用转换选项，您不再需要嵌套分布式 Map 来处理元数据。使用 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 会调用 GetObject API 操作来获取指定的 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"
  },
  ...
]

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

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

在此示例中，假设您有一个名为 factcheck.jsonl 的 JSON Lines 文件。您已将此文件存储在 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"}

分布式 Map 状态可以接受存储在 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 清单文件来指定 Map 状态的数据文件来源。可以将 ManifestType 设置为 ATHENA_DATA，将 InputType 设置为 CSV、JSONL 或 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 是否需要提供列标题。

Map 上下文还将包括 $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 会调用 GetObject API 操作来获取指定的 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 字段语法和传递给子工作流执行的输入的示例。