本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# Step Functions 中的 Map 状态输入和输出字段
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
Map 状态会迭代数据集中的一组项目。数据集的示例包括：
+ 来自先前状态的 JSON 数组和对象。
+ 以 JSON、JSONL、CSV、Parquet 文件等格式存储在 Amazon S3 中的单个数据文件。
+ 对多个对象的引用，例如：Athena 清单和 Amazon S3 清单文件
Map 为数据集中的每个项目重复一组步骤。您可以使用多种配置选项来配置 `Map state` 接收的输入以及 Map 生成的输出。Step Functions 按以下列表所示的顺序应用*分布式 Map 状态*中的每个选项。根据您的使用案例，您可能不需要应用所有字段。
1. [ItemReader （地图）](input-output-itemreader.md) - 用于读取您的数据项
1. [ItemsPath （ JSONPath 仅限地图）](input-output-itemspath.md)或 It **em JSONata s ()**-可选；用于指定数据集中的项目
1. [ItemSelector（Map）](input-output-itemselector.md) - 可选；用于选择和修改数据集中的项目
1. [ItemBatcher（Map）](input-output-itembatcher.md) - 用于在处理大型项目集时处理项目组
1. [ResultWriter （地图）](input-output-resultwriter.md) - 为子工作流的输出结果提供选项
# ItemReader （地图）
`ItemReader` 字段是一个 JSON 对象，用于指定数据集及其位置。*分布式 Map 状态*使用此数据集作为其输入。
以下示例显示了**JSONPath基于**工作流程的`ItemReader`字段语法，适用于存储在 Amazon S3 存储桶中的文本分隔文件中的数据集。
```
"ItemReader": {
"ReaderConfig": {
"InputType": "CSV",
"CSVHeaderLocation": "FIRST_ROW"
},
"Resource": "arn:aws:states:::s3:getObject",
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "csvDataset/ratings.csv",
"VersionId": "BcK42coT2jE1234VHLUvBV1yLNod2OEt"
}
}
```
在**JSONata基于**以下的工作流程中，请注意已替换`Parameters`为**参数**。
```
"ItemReader": {
"ReaderConfig": {
"InputType": "CSV",
"CSVHeaderLocation": "FIRST_ROW"
},
"Resource": "arn:aws:states:::s3:getObject",
"Arguments": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "csvDataset/ratings.csv"
"VersionId": "BcK42coT2jE1234VHLUvBV1yLNod2OEt"
}
}
```
## 该 ItemReader 字段的内容
根据您的数据集，`ItemReader` 字段的内容会有所不同。例如，如果您的数据集是从工作流的上一个步骤传递的 JSON 数组，则 `ItemReader` 字段将被省略。如果您的数据集是 Amazon S3 数据来源，则此字段包含以下子字段。
**`Resource`**
Step Functions 将使用的 Amazon S3 API 集成操作，例如 `arn:aws:states:::s3:getObject`
**`Arguments (JSONata) or Parameters (JSONPath)`**
一个 JSON 对象，用于指定存储数据集的 Amazon S3 存储桶名称和对象密钥。
如果存储桶已启用版本控制，您也可以提供 Amazon S3 对象版本。
**`ReaderConfig`**
一个 JSON 对象，用于指定以下详细信息：
+ `InputType`
接受以下值之一：`CSV`、`JSON`、`JSONL`、`PARQUET`、`MANIFEST`。
指定 Amazon S3 数据来源的类型，例如文本分隔文件 (`CSV`)、对象、JSON 文件、JSON Lines、Parquet 文件、Athena 清单或 Amazon S3 清单列表。在 Workflow Studio 中，您可以从 **S3 项目来源**中选择一种输入类型。
大多数使用 `S3GetObject` 检索的输入类型也支持其参数中的 `ExpectedBucketOwner` 和 `VersionId` 字段。Parquet 文件是唯一不支持 `VersionId` 的类型。
输入文件支持以下外部压缩类型：GZIP、ZSTD。
文件名示例：`myObject.jsonl.gz` 和 `myObject.csv.zstd`。
注意：Parquet 文件是一种内部压缩的二进制文件类型。支持 GZIP、ZSTD 和 Snappy 压缩。
+ `Transformation`
*可选。*值将为 `NONE` 或 `LOAD_AND_FLATTEN`。
如果未指定，则假定为 `NONE`。设置为 `LOAD_AND_FLATTEN` 时，还必须设置 `InputType`。
默认情况下，Map 将对调用 `S3:ListObjectsV2` 时返回的**元数据对象**进行迭代。设置为 `LOAD_AND_FLATTEN` 时，Map 将读取和处理结果列表中引用的实际**数据对象**。
+ `ManifestType`
*可选。*值将为 `ATHENA_DATA` 或 `S3_INVENTORY`。
注意：如果设置为 `S3_INVENTORY`，则**不得**同时指定 `InputType`，因为假定类型为 `CSV`。
+ `CSVDelimiter`
当 `InputType` 为 `CSV` 或 `MANIFEST` 时，您可以指定此字段。
接受以下值之一：`COMMA`（默认）、`PIPE`、`SEMICOLON`、`SPACE`、`TAB`。
**注意**
使用 `CSVDelimiter` 字段，`ItemReader` 可以处理除逗号以外的其他字符分隔的文件。对“CSV 文件”的引用还包括使用 `CSVDelimiter` 字段指定的替代分隔符的文件。
+ `CSVHeaderLocation`
当 `InputType` 为 `CSV` 或 `MANIFEST` 时，您可以指定此字段。
接受以下值之一来指定列标题的位置：
+ `FIRST_ROW` – 如果文件的第一行是标题，则使用此选项。
+ `GIVEN` – 使用此选项在状态机定义中指定标题。
例如，如果您的文件包含以下数据。
```
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"
]
}
}
```
**CSV 标题大小**
对于文本分隔文件，Step Functions 支持的最大标题大小为 10 KiB。
+ `ItemsPointer`
*可选。*您可以在 i `InputType` s 时指定此字段`JSON`。
`ItemsPointer`使用 JSONPointer 语法选择嵌套在 JSON 文件中的特定数组或对象。 JSONPointer 是一种标准化语法，专为在 JSON 文档中导航和引用 JSON 文档中的位置而设计。
JSONPointer syntax 使用正斜杠 (/) 来分隔每个嵌套级别，数组索引表示为不带方括号的数字。例如：
+ `/Data/Contents`-引用 Data 对象中的内容数组
+ `/Data/Contents/0`-引用内容数组的第一个元素
目标数组的起始位置必须在 JSON 文件的前 16MB 以内，并且 JSONPointer 路径的长度必须小于 2000 个字符。
例如，如果您的 JSON 文件包含：
```
{"data": {"items": [{"id": 1}, {"id": 2}]}}
```
你可以指定`"ItemsPointer": "/data/items"`处理项目数组。
+ `MaxItems`
默认情况下，`Map` 状态会迭代指定数据集中的所有项。通过设置 `MaxItems`，您可以限制传递给 `Map` 状态的数据项数量。例如，如果您提供一个包含 1000 行的文本分隔文件，并将限制设置为 100 行，则解释器*仅*向*分布式 Map 状态*传递 100 行。`Map` 状态从标题行之后开始，按顺序处理项。
对于**JSONPath**工作流程，您可以使用`MaxItemsPath`和指向状态输入中解析为整数的键值对的*引用路径*。请注意，您可以指定 `MaxItems` 或 `MaxItemsPath`，但不能同时指定**两者**。
**注意**
您可以将上限指定为 100000000，超过该值后，`Distributed Map` 停止读取数据项。
**账户和区域的相关要求**
您的 Amazon S3 存储桶必须 AWS 账户 与 AWS 区域 您的状态机相同。
*请注意，尽管您的状态机可能能够访问不同存储桶中相同 AWS 区域存储桶中的文件 AWS 账户 ，但 Step Functions 仅支持在 Amazon S3 存储桶中列出与状态机相同 AWS 账户 和 AWS 区域 相同的对象。*
## 处理嵌套数据集（2025 年 9 月 11 日更新）
使用新 `Transformation` 参数，您可以指定值 `LOAD_AND_FLATTEN`，Map 将读取 `S3:ListObjectsV2` 调用结果列表中引用的**实际**数据对象。
在此版本之前，您需要创建嵌套的分布式 Map 来**检索**元数据，然后**处理**实际数据。第一个 Map 将对 `S3:ListObjectsV2` 返回的**元数据**进行迭代，并调用子工作流。每个子状态机中的另一个 Map 将从单个文件中读取**实际数据**。使用转换选项，您可以同时完成这两个步骤。
想象一下，您想对系统每小时生成并存储在 Amazon S3 中的过去 24 个日志文件进行每日审计。您的分布式 Map 状态可以列出带 `S3:ListObjectsV2` 的日志文件，然后迭代每个对象的*元数据*，或者它现在可以加载和分析存储在 Amazon S3 存储桶中的**实际数据**对象。
使用 `LOAD_AND_FLATTEN` 选项可以提高可扩展性，减少打开的 Map Run 次数，并同时处理多个对象。Athena 和 Amazon EMR 作业通常会生成可使用新配置处理的输出。
以下是 `ItemReader` 定义中的参数的示例：
```
{
"QueryLanguage": "JSONata",
"States": {
...
"Map": {
...
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"ReaderConfig": {
// InputType is required if Transformation is LOAD_AND_FLATTEN.
"InputType": "CSV | JSON | JSONL | PARQUET",
// Transformation is OPTIONAL and defaults to NONE if not present
"Transformation": "NONE | LOAD_AND_FLATTEN"
},
"Arguments": {
"Bucket": "amzn-s3-demo-bucket1",
"Prefix": "{% $states.input.PrefixKey %}"
}
},
...
}
}
```
## 数据集示例
您可以指定下列选项之一作为数据集：
+ [上一个步骤中的 JSON 数据](#itemsource-json-array)
+ [Amazon S3 对象列表](#itemsource-example-s3-object-data)
+ [由 LOAD\$1AND\$1FLATTEN 转换的 Amazon S3 对象](#itemsource-example-s3-object-data-flatten)
+ [Amazon S3 存储桶中的 JSON 文件](#itemsource-example-json-data)
+ [Amazon S3 存储桶中的 JSON Lines 文件](#itemsource-example-json-lines-data)
+ [Amazon S3 存储桶中的 CSV 文件](#itemsource-example-csv-data)
+ [Amazon S3 存储桶中的 Parquet 文件](#itemsource-example-parquet-data)
+ [Athena 清单（处理多个项目）](#itemsource-example-athena-manifest-data)
+ [Amazon S3 清单（处理多个项目）](#itemsource-example-s3-inventory)
**注意**
Step Functions 需要适当的权限，才能访问您使用的 Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
### 上一个步骤中的 JSON 数据
*分布式 Map 状态*可以接受从工作流的上一个步骤中传递的 JSON 输入。
输入可以是 JSON 数组、JSON 对象，也可以是 JSON 对象节点内的数组。
Step Functions 将直接迭代数组的元素，或 JSON 对象的键值对。
要从输入中选择包含嵌套 JSON 数组或对象的特定节点，可以在 JSONata 状态`Items`字段中使用`ItemsPath （ JSONPath 仅限地图）`或使用 JSONata 表达式。
要处理单个项目，*分布式 Map 状态*会为每个项目启动子工作流执行。以下选项卡显示了传递给 `Map` 状态的输入以及子工作流执行的相应输入的示例。
**注意**
当数据集是上一个步骤中的 JSON 数据时，不需要 `ItemReader` 字段。
------
#### [ 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 ]
*分布式 Map 状态*将启动三个子工作流执行。每次执行都会接收一个数组项作为输入。以下示例显示了子工作流执行接收的输入。
```
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
}
```
------
### Amazon S3 对象列表
*分布式 Map 状态* 可以迭代存储在 Amazon S3 存储桶中的对象。**当工作流程执行达到`Map`状态时，Step Functions 会调用 [ListObjectsV2](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html) API 操作，该操作会返回 Amazon S3 对象元数据的数组。**在此数组中，每个项目都包含存储在存储桶中的实际数据的数据，例如**ETag**和**密钥**。
要处理数组中的各个项目，*分布式 Map 状态* 会启动一个子工作流执行。例如，假设您的 Amazon S3 存储桶包含 100 张图片。然后，调用 `ListObjectsV2` API 操作后返回的数组包含 100 个元数据项目。然后，*分布式 Map 状态* 将启动 100 个子工作流执行，以处理每个项目。
要在没有嵌套工作流的情况下直接处理数据对象，可以选择 LOAD\$1AND\$1FLATTEN 转换选项来**直接**处理项目。
**注意**
Step Functions 还会为使用 Amazon S3 **控制台**在 Amazon S3 存储桶中创建的每个**文件夹**都提供一个项目。文件夹项目会导致启动额外的子工作流执行。
为避免为每个文件夹创建额外的子工作流程执行，我们建议您使用 AWS CLI 来创建文件夹。有关更多信息，请参阅**《AWS Command Line Interface 用户指南》中的[高级别 Amazon S3 命令](https://docs.aws.amazon.com/cli/latest/userguide/cli-services-s3-commands.html#using-s3-commands-managing-buckets-creating)。
Step Functions 需要适当的权限，才能访问您使用的Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
以下选项卡显示了该数据集的 `ItemReader` 字段语法和传递给子工作流执行的输入的示例。
------
#### [ ItemReader syntax ]
在此示例中，您已将数据（包括图像、JSON 文件和对象）组织在名为 `amzn-s3-demo-bucket` 的 Amazon S3 存储桶中名为 `processData` 的前缀内。
```
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Prefix": "processData"
}
}
```
------
#### [ Input passed to a child workflow execution ]
*分布式 Map 状态*启动的子工作流执行与 Amazon S3 存储桶中存在的元数据项目数量一样多。以下示例显示了子工作流执行接收的输入。
```
{
"Etag": "\"05704fbdccb224cb01c59005bebbad28\"",
"Key": "processData/images/n02085620_1073.jpg",
"LastModified": 1668699881,
"Size": 34910,
"StorageClass": "STANDARD"
}
```
------
### 由 `LOAD_AND_FLATTEN` 转换的 Amazon S3 对象
通过增强对 S3 ListObjects V2 作为分布式地图中的输入源的支持，您的状态机可以直接读取和处理 Amazon S3 存储桶中的多个**数据对象**，无需使用嵌套地图来处理元数据！
使用 `LOAD_AND_FLATTEN` 选项，您的状态机将执行以下操作：
+ 读取 Amazon S3 `ListObjectsV2` 调用列出的每个对象的**实际内容**。
+ 根据 InputType （CSV、JSON、JSONL、Parquet）解析内容。
+ 根据文件内容（行/记录）而不是元数据来创建项目。
使用转换选项，您不再需要嵌套分布式 Map 来处理元数据。使用 LOAD\$1AND\$1FLATTEN 选项可以提高可扩展性，减少活动映射运行次数，并同时处理多个对象。
以下配置显示了 `ItemReader` 的设置：
```
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"ReaderConfig": {
"InputType": "JSON",
"Transformation": "LOAD_AND_FLATTEN"
},
"Arguments": {
"Bucket": "S3_BUCKET_NAME",
"Prefix": "S3_BUCKET_PREFIX"
}
}
```
**存储桶前缀建议**
建议前缀包含尾部斜杠。例如，如果您选择前缀为 `folder1` 的数据，则状态机将处理 `folder1/myData.csv` 和 `folder10/myData.csv`。使用 `folder1/` 时只会处理一个文件夹。
### Amazon S3 存储桶中的 JSON 文件
*分布式 Map 状态* 可以接受存储在 Amazon S3 存储桶中的 JSON 文件作为数据集。JSON 文件必须包含一个数组或 JSON 对象。
当工作流程执行达到`Map`状态时，Step Functions 会调用 [GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html)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 个子工作流执行。
**注意**
用于启动子工作流执行的执行输入不能超过 256 KiB。但是，如果您随后应用可选的 `ItemSelector` 字段来减小项目的大小，Step Functions 支持从文本分隔文件、JSON 或 JSON Lines 文件中读取最大 8 MB 的项目。
Step Functions 支持 Amazon S3 中单个文件的最大大小为 10 GB。
Step Functions 需要适当的权限，才能访问您使用的 Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
以下选项卡显示了该数据集的 `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"
},
...
]
```
------
#### [ ItemReader syntax ]
```
"ItemReader": {
"Resource": "arn:aws:states:::s3:getObject",
"ReaderConfig": {
"InputType": "JSON"
},
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "jsonDataset/factcheck.json"
}
}
```
------
#### [ Input to a child workflow execution ]
*分布式 Map 状态* 启动的子工作流执行与 JSON 文件中存在的数组项数量一样多。以下示例显示了子工作流执行接收的输入。
```
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
}
```
------
### Amazon S3 存储桶中的 JSON Lines 文件
*分布式 Map 状态*可以接受存储在 Amazon S3 存储桶中的 JSON Lines 文件作为数据集。
**注意**
用于启动子工作流执行的执行输入不能超过 256 KiB。但是，如果您随后应用可选的 `ItemSelector` 字段来减小项目的大小，Step Functions 支持从文本分隔文件、JSON 或 JSON Lines 文件中读取最大 8 MB 的项目。
Step Functions 支持 Amazon S3 中单个文件的最大大小为 10 GB。
Step Functions 需要适当的权限，才能访问您使用的 Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
以下选项卡显示了该数据集的 `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"}
```
------
#### [ ItemReader syntax ]
```
"ItemReader": {
"Resource": "arn:aws:states:::s3:getObject",
"ReaderConfig": {
"InputType": "JSONL"
},
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "jsonlDataset/factcheck.jsonl"
}
}
```
------
#### [ Input to a child workflow execution ]
*分布式 Map 状态*启动的子工作流执行与 JSONL 文件中存在的行数一样多。以下示例显示了子工作流执行接收的输入。
```
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
}
```
------
### Amazon S3 存储桶中的 CSV 文件
**注意**
使用 `CSVDelimiter` 字段，`ItemReader` 可以处理除逗号以外的其他字符分隔的文件。对“CSV 文件”的引用还包括使用 `CSVDelimiter` 字段指定的替代分隔符的文件。
*分布式 Map 状态*可以接受存储在 Amazon S3 存储桶中的文本分隔文件作为数据集。如果您使用文本分隔文件作为数据集，则需要指定列标题。有关如何指定标题的信息，请参阅 [该 ItemReader 字段的内容](#itemreader-field-contents)。
Step Functions 根据以下规则解析文本分隔文件：
+ 分隔字段的分隔符由 in 指定。`CSVDelimiter` *ReaderConfig*分隔符默认为 `COMMA`。
+ 换行符是分隔**记录**的分隔符。
+ 字段被视为字符串。对于数据类型转换，使用 [ItemSelector（Map）](input-output-itemselector.md) 中的 `States.StringToJson` 内置函数。
+ 不需要使用双引号（" "）将字符串括起来。但是，用双引号括起来的字符串可以包含逗号和换行符，但不用作记录分隔符。
+ 可以通过重复双引号来保留双引号。
+ 反斜杠（\$1）是另一种转义特殊字符的方法。反斜杠只能与其他反斜杠、双引号以及配置的字段分隔符（如逗号或竖线）配合使用。后跟任何其他字符的反斜杠会被静默移除。
+ 可以通过重复使用反斜杠来保留反斜杠。例如：
```
path,size
C:\\Program Files\\MyApp.exe,6534512
```
+ 转义双引号的反斜杠 (`\"`) 仅在成对出现时才有效，因此建议通过重复使用双引号来转义双引号：`""`。
+ 如果一行中的字段数**少于**标题中的字段数，Step Functions 会为缺失的值提供**空字符串**。
+ 如果一行中的字段数**多于**标题中的字段数，Step Functions 会**跳过**多余的字段。
有关 Step Functions 如何解析文本分隔文件的更多信息，请参阅[Example of parsing an input CSV file](example-csv-parse-dist-map.md#example-csv-parse)。
当工作流程执行达到`Map`状态时，Step Functions 会调[GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html)用 API 操作来获取指定的文件。然后，`Map` 状态会迭代文件中的每一行，并启动一个子工作流执行以处理每行中的项目。例如，假设您提供了一个包含 100 行的文本分隔文件作为输入。然后，解释器将每一行传递给 `Map` 状态。`Map` 状态从标题行之后开始，按顺序处理项目。
**注意**
用于启动子工作流执行的执行输入不能超过 256 KiB。但是，如果您随后应用可选的 `ItemSelector` 字段来减小项目的大小，Step Functions 支持从文本分隔文件、JSON 或 JSON Lines 文件中读取最大 8 MB 的项目。
Step Functions 支持 Amazon S3 中单个文件的最大大小为 10 GB。
Step Functions 需要适当的权限，才能访问您使用的 Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
以下选项卡显示了该数据集的 `ItemReader` 字段语法和传递给子工作流执行的输入的示例。
------
#### [ ItemReader syntax ]
例如，假设您有一个名为 `ratings.csv` 的 CSV 文件。然后，您已将此文件存储在 Amazon S3 存储桶中名为 `csvDataset` 的前缀中。
```
"ItemReader": {
"ReaderConfig": {
"InputType": "CSV",
"CSVHeaderLocation": "FIRST_ROW",
"CSVDelimiter": "PIPE"
},
"Resource": "arn:aws:states:::s3:getObject",
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "csvDataset/ratings.csv"
}
}
```
------
#### [ Input to a child workflow execution ]
*分布式 Map 状态* 启动的子工作流执行与 CSV 文件中存在的行数一样多，不包括标题行（如果文件有）。以下示例显示了子工作流执行接收的输入。
```
{
"rating": "3.5",
"movieId": "307",
"userId": "1",
"timestamp": "1256677221"
}
```
------
### Amazon S3 存储桶中的 Parquet 文件
Parquet 文件可用作输入源。Amazon S3 中存储的 Apache Parquet 文件能够实现大规模的高效列式数据处理。
使用 Parquet 文件时，需满足以下条件：
+ 256MB 是最大行组大小，5MB 是最大页脚大小。如果您提供的输入文件超过任一限制，状态机将返回运行时错误。
+ `InputType=Parquet` **不**支持 `VersionId` 字段。
+ 原生支持内部 GZIP、ZSTD 和 Snappy 数据压缩。无需文件扩展名。
下面显示的是将 `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"
}
}
```
**大规模作业处理**
对于超大规模作业，Step Functions 将使用许多输入读取器。读取器交错进行处理，这可能会导致一些读取器暂停，而另一些读取器继续处理。间歇性进展是大规模作业时的预期行为。
### Athena 清单（处理多个项目）
您可以使用根据 `UNLOAD` 查询结果生成的 Athena 清单文件来指定 Map 状态的数据文件**来源**。可以将 `ManifestType` 设置为 `ATHENA_DATA`，将 `InputType` 设置为 `CSV`、`JSONL` 或 `Parquet`。
运行 `UNLOAD` 查询时，除了实际的数据对象之外，Athena 还会生成一个数据清单文件。清单文件提供了数据文件的结构化 CSV 列表。清单和数据文件都会保存到 Amazon S3 中的 Athena 查询结果位置。
```
UNLOAD () TO 'S3_URI_FOR_STORING_DATA_OBJECT' WITH (format = 'JSON')
```
过程的概念性概述，简要说明如下：
1. 使用 Athena 中的 `UNLOAD` 查询从表中选择数据。
1. Athena 将在 Amazon S3 中生成清单文件（CSV）和数据对象。
1. 将 Step Functions 配置为读取清单文件并处理输入。
该功能可处理来自 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": "",
"Key": "-manifest.csv"
}
}
```
**在 Workflow Studio 中使用 Athena 清单模式**
数据处理的常见场景是将 Map 应用于源自 Athena UNLOAD 查询的数据。Map 会调用 Lambda 函数来处理 Athena 清单中描述的每个项目。Step Functions Workflow Studio 提供了一种现成的模式，可将所有这些组件整合成一个块，您可以将其拖放到状态机画布上。
### S3 清单（处理多个项目）
*分布式 Map 状态* 可以接受存储在 Amazon S3 存储桶中的 Amazon S3 清单文件作为数据集。
当工作流程执行达到`Map`状态时，Step Functions 会调用 [GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html)API 操作来获取指定的 Amazon S3 清单清单文件。
默认情况下，`Map` 状态随后会迭代清单中的**对象**，以返回 Amazon S3 清单对象元数据数组。
如果您指定 ManifestType 为 S3\$1INVENTORY，则 InputType 无法指定。
**注意**
Step Functions 支持解压缩后 Amazon S3 清单报告中单个文件的最大大小为 10 GB。但是，如果每个文件都小于 10 GB，Step Functions 能够处理大小可以超过 10 GB。
Step Functions 需要适当的权限，才能访问您使用的Amazon S3 数据集。有关数据集的 IAM 政策信息，请参阅[关于数据集的 IAM 策略建议](#itemreader-iam-policies)。
以下是 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"
...
```
**重要**
Step Functions 不支持将用户定义的 Amazon S3 清单报告作为数据集。
Amazon S3 清单报告的输出格式必须为 CSV。
有关 Amazon S3 清单及其设置方法的更多信息，请参阅 [Amazon S3 清单](https://docs.aws.amazon.com/AmazonS3/latest/userguide/storage-inventory.html)。
以下 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` 字段语法和传递给子工作流执行的输入的示例。
------
#### [ ItemReader syntax ]
```
"ItemReader": {
"ReaderConfig": {
"InputType": "MANIFEST"
},
"Resource": "arn:aws:states:::s3:getObject",
"Parameters": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Key": "destination-prefix/amzn-s3-demo-bucket/config-id/YYYY-MM-DDTHH-MMZ/manifest.json"
}
}
```
------
#### [ Input to a child workflow execution ]
```
{
"LastModifiedDate": "2022-11-16T00:29:32.000Z",
"Bucket": "amzn-s3-demo-source-bucket",
"Size": "3399671",
"Key": "csvDataset/titles.csv"
}
```
根据您在配置 Amazon S3 清单报告时选择的字段，您的 `manifest.json` 文件内容可能与示例有所不同。
------
## 关于数据集的 IAM 策略建议
当您使用 Step Functions 控制台创建工作流时，Step Functions 可以根据工作流定义中的资源自动生成 IAM 策略。生成的策略包括允许状态机角色调用*分布式地图状态*的 `[StartExecution](https://docs.aws.amazon.com/step-functions/latest/apireference/API_StartExecution.html)` API 操作和访问 AWS 资源（例如 Amazon S3 存储桶和对象以及 Lambda 函数）所需的最低权限。
我们建议在 IAM 策略中仅包括必需的权限。例如，如果您的工作流包含分布式模式下的 `Map` 状态，则将策略范围缩小到包含您的数据的特定 Amazon S3 存储桶和文件夹。
**重要**
如果您在*分布式 Map 状态* 输入中指定了 Amazon S3 存储桶和对象或前缀，并将[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)指向现有键值对，请务必更新工作流程的 IAM 策略。将策略范围缩小到运行时该路径解析到的存储桶和对象名称。
以下示例展示了使用 [ListObjectsV2](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html) 和 [GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html)API 操作授予访问您的 Amazon S3 数据集所需的最低权限的技术。
**Example 使用 Amazon S3 对象作为数据集的条件**
以下条件授予访问 Amazon S3 存储桶的 `processImages` 文件夹中的对象所需的最低权限。
```
"Resource": [ "arn:aws:s3:::amzn-s3-demo-bucket" ],
"Condition": {
"StringLike": {
"s3:prefix": [ "processImages" ]
}
}
```
**Example 使用 CSV 文件作为数据集**
以下示例显示了访问名为 `ratings.csv` 的 CSV 文件所需的操作。
```
"Action": [ "s3:GetObject" ],
"Resource": [
"arn:aws:s3:::amzn-s3-demo-bucket/csvDataset/ratings.csv"
]
```
**Example 使用 Amazon S3 清单作为数据集**
下面显示了 Amazon S3 清单和数据文件的示例资源。
```
"Resource": [
"arn:aws:s3:::myPrefix/amzn-s3-demo-bucket/myConfig-id/YYYY-MM-DDTHH-MMZ/manifest.json",
"arn:aws:s3:::myPrefix/amzn-s3-demo-bucket/myConfig-id/data/*"
]
```
**Example 使用 ListObjects V2 限制使用文件夹前缀**
使用 [ListObjectsV2](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html) 时，将生成两个策略。其中一个策略用于允许**列出**存储桶的内容 (`ListBucket`)，另一个策略将允许**检索存储桶中的对象** (`GetObject`)。
下面显示了示例操作、资源和条件：
```
"Action": [ "s3:ListBucket" ],
"Resource": [ "arn:aws:s3:::amzn-s3-demo-bucket" ],
"Condition": {
"StringLike": {
"s3:prefix": [ "/path/to/your/json/" ]
}
}
```
```
"Action": [ "s3:GetObject" ],
"Resource": [ "arn:aws:s3:::amzn-s3-demo-bucket/path/to/your/json/*" ]
```
请注意，`GetObject` 不会限定范围，您将为对象使用通配符 (`*`)。
# ItemsPath （ JSONPath 仅限地图）
**管理状态和转换数据**
本页指的是 JSONPath。Step Functions 最近添加了变量 JSONata ，用于管理状态和转换数据。
了解如何[使用变量传递数据](workflow-variables.md)和[使用转换数据 JSONata](transforming-data.md)。
在 JSONPath基于状态下，使用该`ItemsPath`字段在提供给`Map`状态的 JSON 输入中选择数组或对象。默认情况下，`Map` 状态将 `ItemsPath` 设置为 `$`，这将选择整个输入。
+ 如果`Map`状态的输入是 JSON 数组，则它会对数组中的每个项目运行迭代，将该项目作为输入传递给迭代
+ 如果`Map`状态的输入是 JSON 对象，则它会为对象中的每个键值对运行迭代，并将该键值对作为输入传递给迭代
**注意**
只有在工作流中使用从上一个状态传递的 JSON 输入时，才能在*分布式 Map 状态*下使用 `ItemsPath`。
的值`ItemsPath`必须是[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)，并且该路径的计算结果必须为 JSON 数组或对象。例如，考虑 `Map` 状态的输入，其中包含两个数组，如以下示例所示。
```
{
"ThingsPiratesSay": [
{
"say": "Avast!"
},
{
"say": "Yar!"
},
{
"say": "Walk the Plank!"
}
],
"ThingsGiantsSay": [
{
"say": "Fee!"
},
{
"say": "Fi!"
},
{
"say": "Fo!"
},
{
"say": "Fum!"
}
]
}
```
在这种情况下，您可以通过 `ItemsPath` 选择一个数组来指定用于 `Map` 状态迭代的数组。以下状态机定义使用 `ItemsPath` 来指定输入中的 `ThingsPiratesSay` 数组，然后对 `ThingsPiratesSay` 数组中的每个项目运行 `SayWord` 传递状态的迭代。
```
{
"StartAt": "PiratesSay",
"States": {
"PiratesSay": {
"Type": "Map",
"ItemsPath": "$.ThingsPiratesSay",
"ItemProcessor": {
"StartAt": "SayWord",
"States": {
"SayWord": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
```
对于嵌套的 JSON 对象`ItemsPath`，您可以使用在输入中选择特定对象。考虑以下带有嵌套配置数据的输入：
```
{
"environment": "production",
"servers": {
"web": {
"server1": {"port": 80, "status": "active"},
"server2": {"port": 8080, "status": "inactive"}
},
"database": {
"primary": {"host": "db1.example.com", "port": 5432},
"replica": {"host": "db2.example.com", "port": 5432}
}
}
}
```
要遍历 Web 服务器对象，您需要设置`ItemsPath`为`$.servers.web`：
```
{
"StartAt": "ProcessWebServers",
"States": {
"ProcessWebServers": {
"Type": "Map",
"ItemsPath": "$.servers.web",
"ItemProcessor": {
"StartAt": "CheckServer",
"States": {
"CheckServer": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
```
处理输入时，`Map` 状态在 [`InputPath`](input-output-inputpath-params.md#input-output-inputpath) 之后应用 `ItemsPath`。`InputPath` 筛选输出后，它将对状态的有效输入进行操作。
有关 `Map` 状态的更多信息，请参阅以下内容：
+ [Map 状态](state-map.md)
+ [Map 状态处理模式](state-map.md#concepts-map-process-modes)
+ [使用内联 Map 状态重复操作](tutorial-map-inline.md)
+ [内联 `Map` 状态输入和输出处理](state-map-inline.md#inline-map-state-output)
# ItemSelector（Map）
**管理状态和转换数据**
了解如何[使用变量在状态之间传递数据](workflow-variables.md)和[使用 JSONata 转换数据](transforming-data.md)。
默认情况下，`Map` 状态的有效输入是原始状态输入中的一组单个数据项。使用 `ItemSelector` 字段，可在数据项的值传递到 `Map` 状态之前对其进行覆盖。
要覆盖这些值，请指定包含键值对集合的有效 JSON 输入。这些键值对可以是状态机定义中提供的静态值，使用[路径](amazon-states-language-paths.md)从状态输入中选择的值，或者从[上下文对象](input-output-contextobject.md)访问的值。
如果您使用路径或上下文对象指定键值对，则键名必须以 `.$` 结尾。
**注意**
`ItemSelector` 字段取代了 `Map` 状态内的 `Parameters` 字段。如果您在 `Map` 状态定义中使用 `Parameters` 字段来创建自定义输入，建议您将其替换为 `ItemSelector`。
您可以在*内联 Map 状态*和*分布式 Map 状态*中指定 `ItemSelector` 字段。
例如，考虑以下 JSON 输入，其中包含 `imageData` 节点内三个项目的数组。对于每次 *`Map` 状态迭代*，都会将一个数组项作为输入传递给迭代。
```
[
{
"resize": "true",
"format": "jpg"
},
{
"resize": "false",
"format": "png"
},
{
"resize": "true",
"format": "jpg"
}
]
```
使用 `ItemSelector` 字段，您可以定义一个自定义 JSON 输入以覆盖原始输入，如以下示例所示。然后，Step Functions 将此自定义输入传递给每次 *`Map` 状态迭代*。自定义输入包含 `size` 的静态值和 `Map` 状态的上下文对象数据的值。`$$.Map.Item.Value` 上下文对象包含每个单独数据项的值。
```
{
"ItemSelector": {
"size": 10,
"value.$": "$$.Map.Item.Value"
}
}
```
以下示例显示了*内联 Map 状态* 的一次迭代所接收的输入：
```
{
"size": 10,
"value": {
"resize": "true",
"format": "jpg"
}
}
```
**提示**
有关使用 `ItemSelector` 字段的*分布式 Map 状态* 的完整示例，请参阅[使用分布式 Map 复制大规模 CSV](tutorial-map-distributed.md)。
# ItemBatcher（Map）
**管理状态和转换数据**
了解如何[使用变量在状态之间传递数据](workflow-variables.md)和[使用 JSONata 转换数据](transforming-data.md)。
`ItemBatcher` 字段是一个 JSON 对象，用于指定在单个子工作流执行中处理一组项目。可在处理大型 CSV 文件或 JSON 数组，或大型 Amazon S3 对象集时，使用批处理。
以下示例显示了 `ItemBatcher` 字段的语法。在以下语法中，每个子工作流执行应处理的最大项目数设置为 100。
```
{
"ItemBatcher": {
"MaxItemsPerBatch": 100
}
}
```
默认情况下，数据集中的每个项目都作为输入传递给单个子工作流执行。例如，假设您指定一个 JSON 文件作为输入，其中包含以下数组：
```
[
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
},
{
"verdict": "false",
"statement_date": "6/7/2022",
"statement_source": "television"
},
{
"verdict": "true",
"statement_date": "5/18/2016",
"statement_source": "news"
},
...
]
```
对于给定的输入，每个子工作流执行都会接收一个数组项作为其输入。以下示例显示了一个子工作流执行的输入：
```
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
}
```
为了帮助优化处理任务的性能和成本，请选择能够平衡项目数量和项目处理时间的批次大小。如果使用批处理，Step Functions 会将这些项目添加到一个**项目**数组中。然后，它将数组作为输入传递给每个子工作流执行。以下示例显示了作为输入传递给子工作流执行的一批两个项目：
```
{
"Items": [
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
},
{
"verdict": "false",
"statement_date": "6/7/2022",
"statement_source": "television"
}
]
}
```
**提示**
要了解有关在工作流中使用 `ItemBatcher` 字段的更多信息，请尝试以下教程和研讨会：
[在 Lambda 函数中处理整批数据](tutorial-itembatcher-param-task.md)
[在子工作流执行中批量迭代项目](tutorial-itembatcher-single-item-process.md)
《The AWS Step Functions Workshop》**中的 [Distributed map and related resources](https://catalog.workshops.aws/stepfunctions/use-cases/distributed-map)
**Contents**
+ [用于指定项目批处理的字段](#input-output-itembatcher-subfields)
## 用于指定项目批处理的字段
要对项目进行批处理，请指定要批处理的最大项目数、最大批量大小，或者同事指定两者。您至少指定一个值才能批处理项目。
**每个批次的最大项目数**
指定每个子工作流执行处理的最大项目数。解释器将 `Items` 数组中批处理的项目数限制为该值。如果您同时指定批处理数量和大小，解释器会减少批次中的项目数以避免超过指定的批次大小限制。
如果您未指定此值但提供了最大批处理大小的值，Step Functions 会在不超过以字节为单位的最大批处理大小的情况下，在每个子工作流执行过程中处理尽可能多的项目。
例如，假设您使用一个包含 1130 个节点的输入 JSON 文件运行执行。如果您为每批指定的最大项目数为 100，Step Functions 将创建 12 个批次。其中，11 个批次各包含 100 个项目，而第十二批包含其余 30 个项目。
或者，您也可以将每个批次的最大项目指定为*分布式 Map 状态*输入中现有键值对的[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)。此路径必须解析为正整数。
例如，给定以下输入：
```
{
"maxBatchItems": 500
}
```
您可以使用参考路径（**仅限 JSONPath**）指定批处理的最大项目数量，如下所示：
```
{
...
"Map": {
"Type": "Map",
"MaxConcurrency": 2000,
"ItemBatcher": {
"MaxItemsPerBatchPath": "$.maxBatchItems"
}
...
...
}
}
```
对于**基于 JSONata** 的状态，您还可以提供计算结果为正整数的 JSONata 表达式。
您可以指定 `MaxItemsPerBatch` 或 `MaxItemsPerBatchPath (JSONPath only)` 子字段，但不能同时指定两者。
**每批最大 KiB 值**
以字节为单位指定批次的最大大小，最大为 256 KiB。如果您同时指定了最大批次数量和大小，Step Functions 会减少批次中的项目数量以避免超过指定的批量大小限制。
或者，您也可以将最大批量大小指定为*分布式地图状态*输入中现有键值对的[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)。此路径必须解析为正整数。
如果您使用批处理但未指定最大批处理大小，则解释器会在每个子工作流执行中尽可能多地处理不超过 256 KiB 的项目。
例如，给定以下输入：
```
{
"batchSize": 131072
}
```
您可以使用参考路径指定最大批次大小，如下所示：
```
{
...
"Map": {
"Type": "Map",
"MaxConcurrency": 2000,
"ItemBatcher": {
"MaxInputBytesPerBatchPath": "$.batchSize"
}
...
...
}
}
```
对于**基于 JSONata** 的状态，您还可以提供计算结果为正整数的 JSONata 表达式。
您可以指定 `MaxInputBytesPerBatch` 或 `MaxInputBytesPerBatchPath`（仅限 JSONPath）子字段，但不能同时指定两者。
**批量输入**
另外，您也可以指定固定的 JSON 输入，将其包含在传递给每个子工作流执行的每个批次中。Step Functions 将此输入与每个子工作流执行的输入合并。例如，给定以下关于项目数组的事实检查日期的固定输入：
```
"ItemBatcher": {
"BatchInput": {
"factCheck": "December 2022"
}
}
```
每个子工作流执行都会收到以下内容作为输入：
```
{
"BatchInput": {
"factCheck": "December 2022"
},
"Items": [
{
"verdict": "true",
"statement_date": "6/11/2008",
"statement_source": "speech"
},
{
"verdict": "false",
"statement_date": "6/7/2022",
"statement_source": "television"
},
...
]
}
```
对于**基于 JSONata** 的状态，您可以直接向 BatchInput 提供 JSONata 表达式，或者在 JSON 对象或数组中使用 JSONata 表达式。
# ResultWriter （地图）
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
`ResultWriter` 字段是一个 JSON 对象，用于为分布式 Map 状态启动的子工作流执行的输出结果提供选项。如果选择导出，可为输出结果指定不同的格式选项，并指定用于存储它们的 Amazon S3 地址。默认情况下，Step Functions 不会导出这些结果。
**Topics**
+ [该 ResultWriter 字段的内容](#input-output-resultwriter-field-contents)
+ [示例](#input-output-resultwriter-examples)
+ [导出到 Amazon S3](#input-output-resultwriter-exporting-to-S3)
+ [适用的 IAM 政策 ResultWriter](#resultwriter-iam-policies)
## 该 ResultWriter 字段的内容
`ResultWriter` 字段包含以下子字段。字段的选择决定了输出的格式以及是否将其导出到 Amazon S3。
**`ResultWriter`**
一个 JSON 对象，用于指定以下详细信息：
+ `Resource`
Step Functions 为导出执行结果而调用的 Amazon S3 API 操作。
+ `Parameters`
一个 JSON 对象，用于指定存储执行输出的 Amazon S3 存储桶名称和前缀。
+ `WriterConfig`
您可以通过此字段配置以下选项。
+ `Transformation`
+ `NONE` - 除了返回工作流元数据外，还原封不动地返回子工作流执行的输出。将子工作流执行结果导出到 Amazon S3 且未指定 `WriterConfig` 时的默认选项。
+ `COMPACT` - 返回子工作流执行的输出。未指定 `ResultWriter` 时的默认选项。
+ `FLATTEN` - 返回子工作流执行的输出。如果子工作流执行输出返回一个数组，则在将结果返回到状态输出或将结果写入 Amazon S3 对象之前，此选项会先对数组进行扁平化。
**注意**
如果子工作流执行失败，Step Functions 会原封不动地返回其执行结果。结果等同于将 `Transformation` 设置为 `NONE`。
+ `OutputType`
+ `JSON` - 将结果格式化为 JSON 数组。
+ `JSONL` - 将结果格式化为 JSON Lines。
**必填字段组合**
`ResultWriter` 字段不能为空。必须至少指定其中一个子字段集。
+ `WriterConfig` - 用于预览格式化后的输出，而不将结果保存到 Amazon S3。
+ `Resource` 和 `Parameters` - 用于将结果保存到 Amazon S3，且无需额外进行格式化。
+ 所有三个字段：`WriterConfig`、`Resource` 和 `Parameters` - 用于格式化输出并将其保存到 Amazon S3。
## 示例配置和转换输出
以下主题演示了 `ResultWriter` 的可能配置设置以及不同转换选项的处理结果示例。
+ [ResultWriter 配置](#input-output-resultwriter-example-configurations)
+ [转换](#input-output-resultwriter-example-transformations)
### ResultWriter 配置示例
以下示例演示了三个字段（`WriterConfig`、`Resources` 和 `Parameters`）可能组合所采用的配置。
**仅 *WriterConfig***
此示例配置了状态输出在预览中的呈现方式，输出格式和转换均在 `WriterConfig` 字段中指定。不存在的 `Resource` 和 `Parameters` 字段（原本应提供 Amazon S3 存储桶规范）表示*状态输出*资源。结果传递到下一个状态。
```
"ResultWriter": {
"WriterConfig": {
"Transformation": "FLATTEN",
"OutputType": "JSON"
}
}
```
**仅限 *Resources* 和 *Parameters***
此示例将状态输出导出到指定的 Amazon S3 存储桶，不会像不存在的 `WriterConfig` 字段所指定的那样进行额外的格式化和转换。
```
"ResultWriter": {
"Resource": "arn:aws:states:::s3:putObject",
"Parameters": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Prefix": "csvProcessJobs"
}
```
**所有三个字段：*WriterConfig*、“*资源*” 和 “*参数”***
此示例会根据 `WriterConfig` 字段中的规范，将状态输出格式化。它还会根据 `Resource` 和 `Parameters` 字段中的规范将其导出到 Amazon S3 存储桶。
```
"ResultWriter": {
"WriterConfig": {
"Transformation": "FLATTEN",
"OutputType": "JSON"
},
"Resource": "arn:aws:states:::s3:putObject",
"Parameters": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Prefix": "csvProcessJobs"
}
}
```
### 转换示例
在这些示例中，假设每个子工作流执行都会返回一个输出，该输出是一个对象数组。
```
[
{
"customer_id": "145538",
"order_id": "100000"
},
{
"customer_id": "898037",
"order_id": "100001"
}
]
```
这些示例演示了不同 `Transformation` 值的格式化输出，其中 `OutputType` 为 `JSON`。
**转换 NONE**
这是使用 `NONE` 转换时的处理结果示例。输出保持不变，并且包含工作流元数据。
```
[
{
"ExecutionArn": "arn:aws:states:region:account-id:execution:orderProcessing/getOrders:da4e9fc7-abab-3b27-9a77-a277e463b709",
"Input": ...,
"InputDetails": {
"Included": true
},
"Name": "da4e9fc7-abab-3b27-9a77-a277e463b709",
"Output": "[{\"customer_id\":\"145538\",\"order_id\":\"100000\"},{\"customer_id\":\"898037\",\"order_id\":\"100001\"}]",
"OutputDetails": {
"Included": true
},
"RedriveCount": 0,
"RedriveStatus": "NOT_REDRIVABLE",
"RedriveStatusReason": "Execution is SUCCEEDED and cannot be redriven",
"StartDate": "2025-02-04T01:49:50.099Z",
"StateMachineArn": "arn:aws:states:region:account-id:stateMachine:orderProcessing/getOrders",
"Status": "SUCCEEDED",
"StopDate": "2025-02-04T01:49:50.163Z"
},
...
{
"ExecutionArn": "arn:aws:states:region:account-id:execution:orderProcessing/getOrders:f43a56f7-d21e-3fe9-a40c-9b9b8d0adf5a",
"Input": ...,
"InputDetails": {
"Included": true
},
"Name": "f43a56f7-d21e-3fe9-a40c-9b9b8d0adf5a",
"Output": "[{\"customer_id\":\"169881\",\"order_id\":\"100005\"},{\"customer_id\":\"797471\",\"order_id\":\"100006\"}]",
"OutputDetails": {
"Included": true
},
"RedriveCount": 0,
"RedriveStatus": "NOT_REDRIVABLE",
"RedriveStatusReason": "Execution is SUCCEEDED and cannot be redriven",
"StartDate": "2025-02-04T01:49:50.135Z",
"StateMachineArn": "arn:aws:states:region:account-id:stateMachine:orderProcessing/getOrders",
"Status": "SUCCEEDED",
"StopDate": "2025-02-04T01:49:50.227Z"
}
]
```
**转换 COMPACT**
这是使用 `COMPACT` 转换时的处理结果示例。请注意，它是具有原始数组结构的子工作流执行的组合输出。
```
[
[
{
"customer_id": "145538",
"order_id": "100000"
},
{
"customer_id": "898037",
"order_id": "100001"
}
],
...,
[
{
"customer_id": "169881",
"order_id": "100005"
},
{
"customer_id": "797471",
"order_id": "100006"
}
]
]
```
**转换 FLATTEN**
这是使用 `FLATTEN` 转换时的处理结果示例。请注意，它是展平为一个数组的子工作流执行数组的组合输出。
```
[
{
"customer_id": "145538",
"order_id": "100000"
},
{
"customer_id": "898037",
"order_id": "100001"
},
...
{
"customer_id": "169881",
"order_id": "100005"
},
{
"customer_id": "797471",
"order_id": "100006"
}
]
```
## 导出到 Amazon S3
**重要**
确保您用于导出 Map Run 结果的 Amazon S3 存储桶AWS 账户与AWS 区域您的状态机相同。否则，您的状态机执行将因 `States.ResultWriterFailed` 错误而失败。
如果您的输出有效载荷大小超过 256 KiB，将结果导出到 Amazon S3 存储桶会很有帮助。Step Functions 整合了所有子工作流执行数据，例如执行输入和输出、ARN 和执行状态。然后，它将状态相同的执行导出到指定 Amazon S3 位置的相应文件中。
以下示例使用**JSONPath**显示了`Parameters`用于导出子工作流程执行结果的`ResultWriter`字段的语法。在此示例中，您将结果存储在名为 `amzn-s3-demo-destination-bucket` 存储桶中，该存储桶位于名为 `csvProcessJobs` 的前缀中。
```
{
"ResultWriter": {
"Resource": "arn:aws:states:::s3:putObject",
"Parameters": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Prefix": "csvProcessJobs"
}
}
}
```
对于**JSONata**州，`Parameters`将替换为`Arguments`。
```
{
"ResultWriter": {
"Resource": "arn:aws:states:::s3:putObject",
"Arguments": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Prefix": "csvProcessJobs"
}
}
}
```
**提示**
在 Workflow Studio 中，您可以通过选择**将 Map 状态结果导出到 Amazon S3** 来导出子工作流执行结果。然后，提供您要将结果导出到其中的 Amazon S3 存储桶的名称和前缀。
Step Functions 需要适当的权限才能访问要导出结果的存储桶和文件夹。有关所需 IAM 策略的信息，请参阅 [适用的 IAM 政策 ResultWriter](#resultwriter-iam-policies)。
如果要导出子工作流执行结果，*分布式 Map 状态*执行将按以下格式返回 Map Run ARN 以及有关 Amazon S3 导出位置的数据：
```
{
"MapRunArn": "arn:aws:states:us-east-2:account-id:mapRun:csvProcess/Map:ad9b5f27-090b-3ac6-9beb-243cd77144a7",
"ResultWriterDetails": {
"Bucket": "amzn-s3-demo-destination-bucket",
"Key": "csvProcessJobs/ad9b5f27-090b-3ac6-9beb-243cd77144a7/manifest.json"
}
}
```
Step Functions 将具有相同状态的执行结果导出到各自的文件中。例如，如果您的子工作流执行结果为 500 个成功和 200 个失败，则 Step Functions 将在指定的 Amazon S3 位置为成功和失败结果创建两个文件。在此示例中，成功结果文件包含 500 个成功结果，而失败结果文件包含 200 个失败结果。
对于给定的执行尝试，Step Functions 会根据您的执行输出在指定的 Amazon S3 位置创建以下文件：
+ `manifest.json` – 包含 Map Run 元数据，例如导出位置、Map Run ARN 以及有关结果文件的信息。
如果您[redriven](redrive-map-run.md)了 Map Run，`manifest.json` 文件会包含 Map Run 的所有尝试中所有成功子工作流执行的引用。但是，此文件包含对特定redrive的失败和待执行的引用。
+ `SUCCEEDED_n.json` – 包含所有成功执行子工作流的合并数据。*n* 表示文件的索引号。索引号从 0 开始。例如 `SUCCEEDED_1.json`。
+ `FAILED_n.json` – 包含所有失败、超时和中止的子工作流执行的合并数据。使用此文件从失败的执行中恢复。*n* 表示文件的索引。索引号从 0 开始。例如 `FAILED_1.json`。
+ `PENDING_n.json` – 包含由于 Map Run 失败或中止而未启动的所有子工作流执行的合并数据。*n* 表示文件的索引。索引号从 0 开始。例如 `PENDING_1.json`。
Step Functions 支持最大为 5 GB 的单个结果文件。如果文件大小超过 5 GB，Step Functions 会创建另一个文件来写入超出部分的执行结果，并在文件名后面附加一个索引号。例如，如果 `SUCCEEDED_0.json` 文件的大小超过 5 GB，Step Functions 会创建一个 `SUCCEEDED_1.json` 文件来记录超出部分结果。
如果您未指定导出子工作流执行结果，则状态机执行将返回子工作流执行结果数组，如以下示例所示：
```
[
{
"statusCode": 200,
"inputReceived": {
"show_id": "s1",
"release_year": "2020",
"rating": "PG-13",
"type": "Movie"
}
},
{
"statusCode": 200,
"inputReceived": {
"show_id": "s2",
"release_year": "2021",
"rating": "TV-MA",
"type": "TV Show"
}
},
...
]
```
**注意**
如果返回的输出大小超过 256 KiB，则状态机执行失败并返回一个 `States.DataLimitExceeded` 错误。
## 适用的 IAM 政策 ResultWriter
当您使用 Step Functions 控制台创建工作流时，Step Functions 可以根据工作流定义中的资源自动生成 IAM 策略。生成的策略包括允许状态机角色调用*分布式地图状态*的 `[StartExecution](https://docs.aws.amazon.com/step-functions/latest/apireference/API_StartExecution.html)` API 操作和访问AWS资源（例如 Amazon S3 存储桶和对象以及 Lambda 函数）所需的最低权限。
我们建议在 IAM 策略中仅包括必需的权限。例如，如果您的工作流包含分布式模式下的 `Map` 状态，则将策略范围缩小到包含您的数据的特定 Amazon S3 存储桶和文件夹。
**重要**
如果您在*分布式 Map 状态* 输入中指定了 Amazon S3 存储桶和对象或前缀，并将[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)指向现有键值对，请务必更新工作流程的 IAM 策略。将策略范围缩小到运行时该路径解析到的存储桶和对象名称。
以下 IAM 策略示例授予使用 `[PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html)` API 操作将子工作流程执行结果写入 Amazon S3 存储桶*csvJobs*中名为的文件夹所需的最低权限。
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:GetObject",
"s3:ListMultipartUploadParts",
"s3:AbortMultipartUpload"
],
"Resource": [
"arn:aws:s3:::amzn-s3-demo-destination-bucket/csvJobs/*"
]
}
]
}
```
如果您正在向其写入子工作流程执行结果的 Amazon S3 存储桶使用AWS Key Management Service(AWS KMS)密钥进行加密，则必须在 IAM 策略中包含必要的AWS KMS权限。有关更多信息，请参阅 [AWS KMS key 加密的 Amazon S3 存储桶的 IAM 权限](iam-policies-eg-dist-map.md#multiupload-dmap-result-policy)。
# Step Functions 如何解析输入 CSV 文件
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
Step Functions 根据以下规则解析文本分隔文件：
+ 分隔字段的分隔符由 in 指定。`CSVDelimiter` *ReaderConfig*分隔符默认为 `COMMA`。
+ 换行符是分隔**记录**的分隔符。
+ 字段被视为字符串。对于数据类型转换，使用 [ItemSelector（Map）](input-output-itemselector.md) 中的 `States.StringToJson` 内置函数。
+ 不需要使用双引号（" "）将字符串括起来。但是，用双引号括起来的字符串可以包含逗号和换行符，但不用作记录分隔符。
+ 可以通过重复双引号来保留双引号。
+ 反斜杠（\$1）是另一种转义特殊字符的方法。反斜杠只能与其他反斜杠、双引号以及配置的字段分隔符（如逗号或竖线）配合使用。后跟任何其他字符的反斜杠会被静默移除。
+ 可以通过重复使用反斜杠来保留反斜杠。例如：
```
path,size
C:\\Program Files\\MyApp.exe,6534512
```
+ 转义双引号的反斜杠 (`\"`) 仅在成对出现时才有效，因此建议通过重复使用双引号来转义双引号：`""`。
+ 如果一行中的字段数**少于**标题中的字段数，Step Functions 会为缺失的值提供**空字符串**。
+ 如果一行中的字段数**多于**标题中的字段数，Step Functions 会**跳过**多余的字段。
**解析输入 CSV 文件的示例**
假设您提供了一个名为 `myCSVInput.csv` 的 CSV 文件，其中包含一行作为输入。然后，您已将此文件存储在名为 `amzn-s3-demo-bucket` 的 Amazon S3 存储桶中。该 CSV 文件如下所示。
```
abc,123,"This string contains commas, a double quotation marks (""), and a newline (
)",{""MyKey"":""MyValue""},"[1,2,3]"
```
以下状态机读取此 CSV 文件，并使用 [ItemSelector（Map）](input-output-itemselector.md)来转换某些字段的数据类型。
```
{
"StartAt": "Map",
"States": {
"Map": {
"Type": "Map",
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "Pass",
"States": {
"Pass": {
"Type": "Pass",
"End": true
}
}
},
"End": true,
"Label": "Map",
"MaxConcurrency": 1000,
"ItemReader": {
"Resource": "arn:aws:states:::s3:getObject",
"ReaderConfig": {
"InputType": "CSV",
"CSVHeaderLocation": "GIVEN",
"CSVHeaders": [
"MyLetters",
"MyNumbers",
"MyString",
"MyObject",
"MyArray"
]
},
"Parameters": {
"Bucket": "amzn-s3-demo-bucket",
"Key": "myCSVInput.csv"
}
},
"ItemSelector": {
"MyLetters.$": "$$.Map.Item.Value.MyLetters",
"MyNumbers.$": "States.StringToJson($$.Map.Item.Value.MyNumbers)",
"MyString.$": "$$.Map.Item.Value.MyString",
"MyObject.$": "States.StringToJson($$.Map.Item.Value.MyObject)",
"MyArray.$": "States.StringToJson($$.Map.Item.Value.MyArray)"
}
}
}
}
```
当您运行这个状态机时，它会产生以下输出。
```
[
{
"MyNumbers": 123,
"MyObject": {
"MyKey": "MyValue"
},
"MyString": "This string contains commas, a double quote (\"), and a newline (\n)",
"MyLetters": "abc",
"MyArray": [
1,
2,
3
]
}
]
```