ItemReader - Amazon Step Functions
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

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

ItemReader

ItemReader字段是一个 JSON 对象,用于指定数据集及其位置。一个分布式地图状态使用此数据集作为其输入。以下示例显示语法ItemReader字段,前提是您的数据桶中存储桶中可使用。

"ItemReader": { "ReaderConfig": { "InputType": "CSV", "CSVHeaderLocation": "FIRST_ROW" }, "Resource": "arn:aws:states:::s3:getObject", "Parameters": { "Bucket": "myBucket", "Key": "csvDataset/ratings.csv" } }
提示

在 Workflow Studio 中,您可以指定数据集及其在物品来源字段。

的内容 ItemReader 领域

根据您的数据集,其中的内容ItemReader字段各不相同。例如,如果您的数据集是从工作流程的上一步传递的 JSON 数组,则ItemReader字段被省定。如果您的数据集是 Amazon S3 数据源,则此字段包含以下子字段。

ReaderConfig

指定以下详细信息的 JSON 对象:

  • InputType

    指定 Amazon S3 数据源的类型,例如 CSV 文件、对象、JSON 文件或亚马逊 S3 清单清单。在 Workflow Studio 中,你可以从中选择一种输入类型Amazon S3 商品来源下拉列表物品来源字段。

  • CSVHeaderLocation

    注意

    仅当使用 CSV 文件作为数据集时,才必须指定此字段。

    接受以下值之一来指定列标题的位置:

    重要

    目前,Step Functions 支持最大 10 KB 的 CSV 标头。

    • FIRST_ROW— 如果文件的第一行是标题,则使用此选项。

    • GIVEN— 使用此选项在状态机定义中指定标题。例如,假定 CSV 文件包含以下数据。

      1,307,3.5,1256677221 1,481,3.5,1256677456 1,1091,1.5,1256677471 ...

      提供以下 JSON 数组作为 CSV 标题。

      "ItemReader": { "ReaderConfig": { "InputType": "CSV", "CSVHeaderLocation": "GIVEN", "CSVHeaders": [ "userId", "movieId", "rating", "timestamp" ] } }
    提示

    在 Workflow Studio 中,你可以在其他配置物品来源字段。

  • MaxItems

    限制传递给数据项的数量Map州。例如,假设您提供的一个包含 1000 行的 CSV 文件并指定限制为 100。然后,口译员通过只有100 行到了Map州。的Mapstate 按顺序处理项目,从标题行之后开始。

    默认情况下,Mapstate 迭代指定数据集中的所有项目。

    注意

    目前,您可以将上限指定为1亿英镑。的分布式地图状态停止读取超过此限制的项目。

    提示

    在 Workflow Studio 中,你可以在其他配置物品来源字段。

    或者,您可以指定参考路径到您现有的键值对分布式地图状态输入。此路径必须解析为正整数。您可以在中指定参考路径MaxItemsPath子字段。

    重要

    您可以指定MaxItems或者MaxItemsPath子字段,但不能两者兼而有之。

Resource

Step Functions 必须根据指定的数据集调用的 Amazon S3 API 操作。

Parameters

一个 JSON 对象,用于指定存储数据集的 Amazon S3 存储桶名称和对象密钥。

重要

确保您的 Amazon S3 存储桶处于相同位置Amazon Web Services 账户和Amazon Web Services 区域作为您的状态机。

数据集示例

您可以指定以下选项之一作为数据集:

重要

Step Functions 需要适当的权限才能访问您使用的Amazon S3 数据集。有关数据集 IAM 策略的信息,请参阅适用于数据集的 IAM 策略

一个分布式地图状态可以接受工作流程中上一步中传递的 JSON 输入。此输入必须是数组,或者必须包含特定节点内的数组。要选择包含阵列的节点,可以使用ItemsPath字段。

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

注意

Step Functions 省略了ItemReader当您的数据集是上一步中的 JSON 数组时为字段。

Input passed to the Map state

考虑以下由三个项目组成的 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" } ]
Input passed to a child workflow execution

分布式地图状态启动三个子工作流程执行。每次执行都会接收一个数组项作为输入。以下示例显示了子工作流程执行收到的输入。

{ "verdict": "true", "statement_date": "6/11/2008", "statement_source": "speech" }

一个分布式地图状态可以遍设存储在 Amazon S3 存储桶中的对象。当工作流程执行到达Map状态,Step Functions 调用ListObjectsV2API 操作,它返回 Amazon S3 对象元数据的数组。在这个数组中,每个项目都包含数据,例如etag钥匙,用于存储在存储桶中的数据。

要处理数组中的单个项目,分布式地图状态启动子工作流程执行。例如,假定 Amazon S3 存储桶拥有图片。然后,数组在调用后返回ListObjectsV2API 操作包含 100 个项目。的分布式地图状态然后启动 100 个子工作流程执行来处理每个数组项目。

注意
  • 目前,Step Functions 还为您使用 Amazon S3 存储桶创建的项目可使用 Amazon S3 存储桶创建的文件夹。这会导致额外的子工作流程执行由分布式地图状态。为避免为该文件夹创建额外的子工作流程执行,我们建议您使用Amazon CLI来创建文件夹。有关更多信息,请参阅高级亚马逊 S3 命令Amazon Command Line Interface用户指南

  • Step Functions 需要适当的权限才能访问您使用的Amazon S3 数据集。有关数据集 IAM 策略的信息,请参阅适用于数据集的 IAM 策略

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

ItemReader syntax

在此示例中,您已将数据(包括图像、JSON 文件和对象)组织在名为的前缀中processData在名为 Amazon S3 存储桶中myBucket

"ItemReader": { "Resource": "arn:aws:states:::s3:listObjectsV2", "Parameters": { "Bucket": "myBucket", "Prefix": "processData" } }
Input passed to a child workflow execution

分布式地图状态启动的子工作流程执行次数与 Amazon S3 存储桶中存在的项目数一样多。以下示例显示了子工作流程执行收到的输入。

{ "Etag": "\"05704fbdccb224cb01c59005bebbad28\"", "Key": "processData/images/n02085620_1073.jpg", "LastModified": 1668699881, "Size": 34910, "StorageClass": "STANDARD" }

一个分布式地图状态可以接受存储在 Amazon S3 存储桶中的 JSON 文件中。JSON 文件必须包含一个数组。

当工作流程执行到达Map状态,Step Functions 调用GetObject用于获取指定的 JSON 文件的 API 操作。的Map然后,state 遍历数组中的每个项目,并开始对每个项目执行子工作流程。例如,如果您的 JSON 文件包含 1000 个数组项目,则Map状态启动 1000 个子工作流程执行。

注意
  • 用于启动子工作流程执行的执行输入不能超过 256 KB。但是,如果您随后应用可选的,Step Functions 支持从 CSV 或 JSON 文件中读取最大 8 MB 的项目ItemSelector字段可减小项目的大小。

  • 目前,Step Functions 支持将 Amazon S3 库存报告中单个文件的最大大小为 10 GB。但是,如果每个文件都小于 10 GB,Step Functions 可以处理超过 10 GB 的容量。

  • Step Functions 需要适当的权限才能访问您使用的Amazon S3 数据集。有关数据集 IAM 策略的信息,请参阅适用于数据集的 IAM 策略

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

在此示例中,假定 JSON 文件名为factcheck.json。您已将此文件存储在名为的前缀中jsonDataset在 Amazon S3 存储桶中。以下是 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" }, ... ]
ItemReader syntax
"ItemReader": { "Resource": "arn:aws:states:::s3:getObject", "ReaderConfig": { "InputType": "JSON" }, "Parameters": { "Bucket": "myBucket", "Key": "jsonDataset/factcheck.json" } }
Input to a child workflow execution

分布式地图状态启动的子工作流程执行次数与 JSON 文件中存在的数组项数量一样多。以下示例显示了子工作流程执行收到的输入。

{ "verdict": "true", "statement_date": "6/11/2008", "statement_source": "speech" }

一个分布式地图状态可以接受存储在 Amazon S3 存储桶中存储桶中的 CSV 文件可使用。如果您使用 CSV 文件作为数据集,则需要指定 CSV 列标题。有关如何指定 CSV 标头的信息,请参见的内容 ItemReader 领域

由于在 CSV 文件中创建和维护 CSV 文件中没有标准化格式,因此 Step Functions 会根据以下规则解析 CSV 文件:

  • 逗号 (,) 是分隔各个字段的分隔符。

  • 换行符是分隔各个记录的分隔符。

  • 字段被视为字符串。要进行数据类型转换,请使用States.StringToJson内在函数ItemSelector

  • 字符串不需要双引号 (” “)。但是,用双引号括起来的字符串可以包含逗号和换行符,但不能用作分隔符。

  • 通过重复双引号来避免双引号。

  • 如果一行中的字段数少于标题中的字段数,Step Functions 会为缺失值提供空字符串。

  • 如果一行中的字段数大于标题中的字段数,Step Functions 会跳过其他字段。

有关 Step Functions 如何解析 CSV 文件的信息,请参阅Example of parsing an input CSV file

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

注意
  • 用于启动子工作流程执行的执行输入不能超过 256 KB。但是,如果您随后应用可选的,Step Functions 支持从 CSV 或 JSON 文件中读取最大 8 MB 的项目ItemSelector字段可减小项目的大小。

  • 目前,Step Functions 支持将 Amazon S3 库存报告中单个文件的最大大小为 10 GB。但是,如果每个文件都小于 10 GB,Step Functions 可以处理超过 10 GB 的容量。

  • Step Functions 需要适当的权限才能访问您使用的Amazon S3 数据集。有关数据集 IAM 策略的信息,请参阅适用于数据集的 IAM 策略

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

ItemReader syntax

例如,假定 CSV 文件名为ratings.csv。然后,您已将此文件存储在名为的前缀中csvDataset在 Amazon S3 存储桶中。

{ "ItemReader": { "ReaderConfig": { "InputType": "CSV", "CSVHeaderLocation": "FIRST_ROW" }, "Resource": "arn:aws:states:::s3:getObject", "Parameters": { "Bucket": "myBucket", "Key": "csvDataset/ratings.csv" } } }
Input to a child workflow execution

分布式地图状态启动的子工作流程执行次数与 CSV 文件中存在的行数一样多,不包括标题行(如果在文件中)。以下示例显示了子工作流程执行收到的输入。

{ "rating": "3.5", "movieId": "307", "userId": "1", "timestamp": "1256677221" }

一个分布式地图状态可以接受存储在 Amazon S3 存储桶中存储桶中的 Amazon S3 清单文件可使用。

当工作流程执行到达Map状态,Step Functions 调用GetObject用于获取指定的 Amazon S3 库存清单文件的 API 操作。的Map然后,state 会遍历清单中的对象,返回一组 Amazon S3 清单对象元数据。

注意
  • 目前,Step Functions 支持将 Amazon S3 库存报告中单个文件的最大大小为 10 GB。但是,如果每个文件都小于 10 GB,Step Functions 可以处理超过 10 GB 的容量。

  • Step Functions 需要适当的权限才能访问您使用的Amazon S3 数据集。有关数据集 IAM 策略的信息,请参阅适用于数据集的 IAM 策略

以下 CSV 格式库存文件示例。此文件包含名为的对象csvDatasetimageDataset,它们存储在名为 Amazon S3 存储桶中sourceBucket

"sourceBucket","csvDataset/","0","2022-11-16T00:27:19.000Z" "sourceBucket","csvDataset/titles.csv","3399671","2022-11-16T00:29:32.000Z" "sourceBucket","imageDataset/","0","2022-11-15T20:00:44.000Z" "sourceBucket","imageDataset/n02085620_10074.jpg","27034","2022-11-15T20:02:16.000Z" ...
重要

目前,Step Functions 不支持将用户定义的 Amazon S3 库存报告作为数据集。您还必须确保 Amazon S3 库存报告的输出格式为 CSV。有关 Amazon S3 库存及其设置方法的更多信息,请参阅Amazon S3 清单Amazon S3 用户指南

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

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

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

ItemReader syntax
{ "ItemReader": { "ReaderConfig": { "InputType": "MANIFEST" }, "Resource": "arn:aws:states:::s3:getObject", "Parameters": { "Bucket": "destinationBucket", "Key": "destination-prefix/source-bucket/config-ID/YYYY-MM-DDTHH-MMZ/manifest.json" } } }
Input to a child workflow execution
{ "LastModifiedDate": "2022-11-16T00:29:32.000Z", "Bucket": "sourceBucket", "Size": "3399671", "Key": "csvDataset/titles.csv" }

根据您在配置 Amazon S3 库存报告时选择的字段,您的内容manifest.json文件可能与显示的示例有所不同。

适用于数据集的 IAM 策略

当您使用 Step Functions 控制台创建工作流程时,Step Functions 可以根据工作流程定义中的资源自动生成 IAM 策略。这些策略包括允许状态机角色调用StartExecution适用于 API 操作分布式地图状态。这些策略还包括 Step Functions 访问所需的最低权限Amazon资源,例如 Amazon S3 存储桶和对象以及。我们强烈建议您在 IAM 策略中仅包含必要的权限。例如,如果您的工作流程包含Map在分布式模式下,将您的策略范围缩小到包含您的数据集的特定 Amazon S3 存储桶和文件夹。

重要

如果您指定 Amazon S3 存储桶和对象(或前缀)参考路径到您现有的键值对分布式地图状态输入,请确保您更新工作流程的 IAM 策略。将策略范围缩小到运行时路径解析到的存储桶和对象名称。

以下 IAM 策略示例授予使用访问您的 Amazon S3 数据集所需的最低权限ListObjectsV2GetObjectAPI 操作。

例 Amazon S3 对象作为数据集的 IAM 策略

以下示例显示了一个 IAM 策略,该策略授予访问其中组织的对象的最低权限processImages在名为 Amazon S3 存储桶中myBucket

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::myBucket" ], "Condition": { "StringLike": { "s3:prefix": [ "processImages" ] } } } ] }
例 将 CSV 文件作为数据集的 IAM 政策

以下示例显示了一个 IAM 策略,该策略授予访问名为的 CSV 文件的最低权限ratings.csv

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": [ "arn:aws:s3:::myBucket/csvDataset/ratings.csv" ] } ] }
例 以数据集形式设Amazon S3 清单的 IAM 政策

以下示例显示了授予访问 Amazon S3 库存报告的最低权限的 IAM 策略。

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": [ "arn:aws:s3:::destination-prefix/source-bucket/config-ID/YYYY-MM-DDTHH-MMZ/manifest.json", "arn:aws:s3:::destination-prefix/source-bucket/config-ID/data/*" ] } ] }