本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 在 Step Functions 中处理输入和输出
**使用变量管理状态和 JSONata**
Step Functions 最近添加了变量 JSONata ，用于管理状态和转换数据。
要了解更多信息，请阅读博客文章 “[使用变量简化开发者体验 JSONata ” 和 AWS Step Functions](https://aws.amazon.com/blogs/compute/simplifying-developer-experience-with-variables-and-jsonata-in-aws-step-functions/)
当 Step Functions 执行收到 JSON 输入时，它会将该数据传递到工作流中的第一个状态作为输入。
使用 JSONata，您可以从中检索状态输入`$states.input`。您的状态机执行还会在 [Context 对象](input-output-contextobject.md) 中提供该初始输入数据。您可以在工作流中的任意步骤从 `$states.context.Execution.Input` 中检索原始状态机输入。
当状态退出时，其输出可用于状态机中的*下一个*状态。除非您**修改**状态输出，否则默认情况下，您的状态输入将作为状态输出来传递。对于后续步骤中可能需要的数据，可以考虑将其存储在变量中。有关更多信息，请参阅[使用变量在状态之间传递数据](workflow-variables.md)。
**QueryLanguage 推荐**
对于新的状态机，我们建议使用 JSONata 查询语言。在未指定查询语言的状态机中，为了 JSONPath 向后兼容，状态机默认为。您必须选择使用 JSONata 您的状态机或单个状态。
**使用处理输入和输出 JSONata**
使用 JSONata 表达式，您可以选择和转换数据。在 `Arguments` 字段中，您可以自定义发送给操作的数据。结果可以在 `Output` 字段中转换为自定义状态输出。您也可以将数据存储在 `Assign` 字段的变量中。有关更多信息，请参阅[使用转换数据 JSONata](transforming-data.md)。
下图显示了 JSON 信息如何在 JSONata 任务状态中移动。
![\[Diagram showing JSONata task state flow with input, arguments, output, and action components.\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-jsonata.png)
**使用处理输入和输出 JSONPath**
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
对于使用的状态机 JSONPath，以下字段控制从一个状态到另一个状态的数据流：`InputPath``Parameters`、`ResultSelector`、`ResultPath`、和`OutputPath`。每个 JSONPath 字段都可以在 JSON 在工作流程中的每个状态中移动时对其进行操作。
JSONPath 字段可以使用[路径](amazon-states-language-paths.md)从输入或结果中选择 JSON 的各个部分。路径是以 `$` 开头的字符串，标识 JSON 文本内的节点。Step Functions 路径使用[JsonPath](https://datatracker.ietf.org/wg/jsonpath/about/)语法。
下图显示了 JSON 信息如何在 JSONPath 任务状态中移动。`InputPath`选择要传递给`Task`状态任务的 JSON 输入部分（例如，AWS Lambda函数）。您可以调整在 `Parameters` 字段中发送给操作的数据。然后，借助 `ResultSelector`，您可以选择要转移的操作结果部分。`ResultPath` 然后选择要传递给输出的状态输入和任务结果的组合。`OutputPath` 可以筛选 JSON 输出以进一步限制传递给输出的信息。
![\[筛选器顺序： InputPath、参数 ResultSelector、 ResultPath、和 OutputPath。\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-jsonpath.png)
**Topics**
+ [使用变量在状态之间传递数据](workflow-variables.md)
+ [在 Step Functi JSONata ons 中使用转换数据](transforming-data.md)
+ [在 Step Functions 中从上下文对象访问执行数据](input-output-contextobject.md)
+ [使用 JSONPath 路径](amazon-states-language-paths.md)
+ [在 Step Functions 工作流中操作参数](input-output-inputpath-params.md)
+ [示例：在 Step Functions 工作流中操作带路径的状态数据](input-output-example.md)
+ [在 Step Functions ResultPath 中使用指定状态输出](input-output-resultpath.md)
+ [Step Functions 中的 Map 状态输入和输出字段](input-output-fields-dist-map.md)
# 使用变量在状态之间传递数据
**使用变量和 JSONata 管理状态**
Step Functions 最近添加了变量和 JSONata 来管理状态和转换数据。
要了解更多信息，请阅读博客文章[使用 AWS Step Functions 中的变量和 JSONata 简化开发人员体验](https://aws.amazon.com/blogs/compute/simplifying-developer-experience-with-variables-and-jsonata-in-aws-step-functions/)
以下视频以 DynamoDB 为例，描述了 Step Functions 中的变量和 JSONata：
使用变量和状态输出，您可以在工作流的各个步骤之间传递数据。
使用工作流变量，您可以在一个步骤中存储数据，并在将来的步骤中检索该数据。例如，您可以存储包含日后可能需要的数据的 API 响应。相反，状态输出只能用作紧接着下一步的输入。
## 变量的概念性概述
使用工作流变量，您可以存储数据以备日后参考。例如，步骤 1 可能会存储 API 请求的结果，以便之后在步骤 5 中重复使用该请求的一部分。
在以下场景中，状态机从 API 获取数据一次。在步骤 1 中，工作流将返回的 API 数据（每个状态最多 256 KiB）存储在变量“x”中，以便在后续步骤中使用。
如果不使用变量，则需要通过输出将该数据从步骤 1 依次传递到步骤 2、步骤 3、步骤 4，然后在步骤 5 中使用。如果这些中间步骤不需要该数据，怎么办？ 通过输出和输入在状态之间传递数据将不必要。
借助变量，您可以存储数据并在将来的任何步骤中使用这些数据。您还可以在不中断数据流的情况下修改、重新排列或添加步骤。考虑到变量的灵活性，您可能只需使用 **Output** 来返回 Parallel 和 Map 子工作流中的数据，并在状态机执行结束时返回。
![\[Diagram showing step 1 assigning a value to $x, used in step 5.\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-diag-opt1.png)
**支持变量的状态**
以下状态类型支持 `Assign` 以声明变量并为其赋值：*Pass、Task、Map、Parallel、Choice、Wait。*
要设置变量，请提供一个包含变量名称和值的 JSON 对象：
```
"Assign": {
"productName": "product1",
"count" : 42,
"available" : true
}
```
要引用变量，请在名称前面加上美元符号（`$`），例如 `$productName`。
## 保留变量：\$1states
Step Functions 定义了一个名为 **`$states`** 的保留变量。在 JSONata 状态下，将以下结构分配给 `$states` 以在 JSONata 表达式中使用：
```
# Reserved $states variable in JSONata states
$states = {
"input": // Original input to the state
"result": // API or sub-workflow's result (if successful)
"errorOutput": // Error Output (only available in a Catch)
"context": // Context object
}
```
进入状态时，Step Functions 会将状态输入分配给 **`$states.input`**。`$states.input` 的值可用于所有接受 JSONata 表达式的字段。`$states.input` 始终是指原始状态输入。
对于 `Task`、`Parallel` 和 `Map` 状态：
+ **`$states.result`** 指 API 或子工作流成功时的原始结果。
+ **`$states.errorOutput`** 指 API 或子工作流失败时的错误输出。
`$states.errorOutput` 可用于 `Catch` 字段的 `Assign` 或 `Output`。
如果在无法访问 `$states.result` 或 `$states.errorOutput` 的字段和状态中尝试对其进行访问，则在创建、更新或验证状态机时将捕获该尝试。
`$states.context` 对象为您的工作流提供有关其具体执行的信息，例如 `StartTime`、任务令牌和初始工作流输入。要了解更多信息，请参阅[在 Step Functions 中从上下文对象访问执行数据](input-output-contextobject.md)。
## 变量名称语法
变量名称遵循 [Unicode® 标准附录 \$131](https://unicode.org/reports/tr31/) 中所述的 Unicode 标识符规则。变量名称的第一个字符必须是 Unicode ID\$1Start 字符，第二个及后续字符必须是 Unicode ID\$1Continue 字符。变量名称的最大长度为 80 个字符。
变量名称约定类似于 JavaScript 和其他编程语言的规则。
## 变量作用域
Step Functions 工作流程通过使用 *workflow-local* 作用域来避免变量出现竞争条件。
workflow-local 作用域包括状态机**状态**字段内的所有状态，但不包括 Parallel 或 Map 状态内的状态。Parallel 或 Map 状态内的状态可以引用外部作用域变量，但它们创建和维护自己的独立 workflow-local 变量和值。
`Parallel` 分支和 `Map` 迭代可以访问**外部作用域**中的变量值，但它们无法访问其他并发分支或迭代中的变量值。处理错误时，`Catch` 中的 `Assign` 字段可以为外部作用域（即 Parallel/Map 状态存在的作用域）中的变量赋值。
异常：**分布式 Map 状态**当前不能引用外部作用域中的变量。
如果作用域中的任何状态为变量分配值，则该变量存在于作用域中。为避免常见错误，内部作用域中分配的变量不能与外部作用域中分配的变量同名。例如，如果顶级作用域为名为 `myVariable` 的变量赋值，则任何其他作用域（在 `Map`、`Parallel` 内）都不能同时为 `myVariable` 赋值。
对变量的访问取决于当前作用域。Parallel 和 Map 状态都有各自的作用域，但它们能够访问外部作用域中的变量。
当 Parallel 或 Map 状态完成时，它们的所有变量都将脱离作用域，不再可访问。使用 **Output 字段**将数据从 Parallel 分支和 Map 迭代中传出。
## 分配 ASL 中的字段
ASL 中的 `Assign` 字段用于为一个或多个变量赋值。`Assign` 字段位于每个状态（`Succeed` 和 `Fail` 除外）的顶级、`Choice` 状态规则内部和 `Catch` 字段内部。例如：
```
# Example of Assign with JSONata
"Store inputs": {
"Type": "Pass",
"Next": "Get Current Price",
"Comment": "Store the input desired price into a variable: $desiredPrice",
"Assign": {
"desiredPrice": "{% $states.input.desired_price %}",
"maximumWait": "{% $states.input.max_days %}"
}
},
```
`Assign` 字段采用一个 JSON 对象。每个顶级字段都命名一个要赋值的变量。在前面的示例中，变量名称为 `desiredPrice` 和 `maximumWait`。使用 JSONata 时，`{% ... %}` 表示可能包含变量或更复杂表达式的 JSONata 表达式。有关 JSONata 表达式的更多信息，请参阅 [JSONata.org 文档](https://docs.jsonata.org/overview.html)。
使用 **JSONata** 作为查询语言时，下图显示了如何并行处理 **Assign** 和 **Output** 字段。注意其含义：*为变量赋值并不会影响状态输出*。
![\[Diagram showing a comparison of JSONPath and JSONata flow.\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-jsonata.png)
以下 JSONata 示例从状态输入中检索 `order.product`。变量 `currentPrice` 设置为任务结果中的一个值。
```
# Example of Task with JSONata assignment from result
{
"Type": "Task",
...
"Assign": {
"product": "{% $states.input.order.product %}",
"currentPrice": "{% $states.result.Payload.current_price %}"
},
"Next": "the next state"
}
```
注意：**不能**为变量的一部分赋值。例如，可以 `"Assign":{"x":42}`，但不能 `"Assign":{"x.y":42}` 或 `"Assign":{"x[2]":42}`。
## 分配字段中的求值顺序
Step Functions 状态中的所有变量引用都使用**状态输入**时的值。
前述事实对于了解 `Assign` 字段如何为一个或多个变量赋值非常重要。首先，计算新值，然后 Step Functions 将新值分配给变量。新的变量值将在**下一个**状态开始时生效。例如，考虑以下 `Assign` 字段：
```
# Starting values: $x=3, $a=6
"Assign": {
"x": "{% $a %}",
"nextX": "{% $x %}"
}
# Ending values: $x=6, $nextX=3
```
在前述示例中，变量 `x` 既被赋值，又被引用。
请记住，所有表达式都是***先求值***，再进行赋值。新分配的值将在**下一个**状态下生效。
我们来详细看一下这个例子。假设在之前的状态下，为 `$x` 分配的值为三（3），为 `$a` 分配的值为六（6）。以下步骤描述了此过程：
1. 使用所有变量的**当前**值计算所有表达式。
表达式 `"{% $a %}"` 的计算结果将为 6，`"{% $x %}"` 的计算结果为 3。
1. 接下来，进行赋值：
将为 `$x` 分配六（6）的值
将为 `$nextX` 分配三（3）的值
注意：如果 `$x` 之前未分配过，则该示例将**失败**，因为 `$x` 将是*未定义*。
总而言之，Step Functions 会先计算**所有**表达式，再进行赋值。变量在 `Assign` 字段中出现的顺序并**不**重要。
## 限制
对于标准和快速工作流，单个变量的最大大小为 256Kib。
单个 `Assign` 字段中所有变量的最大组合大小也为 256Kib。例如，您可以将 X 和 Y 分配为 128KiB 的大小，但不能在同一 `Assign` 字段中将 X 和 Y 都分配为 256KiB 的大小。
每次执行时，所有已存储变量的总大小不能超过 10MiB。
## 在 JSONPath 状态下使用变量
在使用 JSONPath 作为查询语言的状态下，变量也可用。
您可以在任何接受 JSONPath 表达式（`$.` 或 `$$.` 语法）的字段中引用变量，但 `ResultPath` 字段除外，该字段在状态输入中指定用于注入状态结果的位置。不能在 `ResultPath` 中使用变量。
在 JSONPath 中，`$` 符号指的是“当前”值，`$$` 表示状态上下文对象。JSONPath 表达式可以像在 `$.customer.name` 中那样以 `$.` 开头。您可以像在 `$$.Execution.Id` 中那样使用 `$$.` 访问上下文。
要引用变量，也可以在变量名前使用 `$` 符号，例如 `$x` 或 `$order.numItems`。
例如，在接受内置函数的 **JSONPath** 字段中，可以在参数中使用变量，例如 `States.Format('The order number is {}', $order.number)`。
下图说明了 **JSONPath** 任务中的分配步骤是如何与 ResultSelector 同时发生：
![\[Logical diagram of a state that uses JSONPath query language.\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-jsonpath.png)
**在 JSONPath 中分配变量**
JSONPath 变量分配的行为与有效载荷模板类似。以 `.$` 结尾的字段表示该值是一个 JSONPath 表达式，Step Functions 在状态机执行期间将其计算为一个值（例如：`$.order..product` 和 `$.order.total`）。
```
# Example of Assign with JSONPath
{
"Type": "Task",
...
"Assign": {
"products.$": "$.order..product",
"orderTotal.$": "$.order.total"
},
"Next": "the next state"
}
```
对于 JSONPath 状态，`Assign` 字段中 `$` 的值取决于状态类型。在 `Task,`、`Map`、`Parallel` 状态下，`$` 指的是 API/子工作流结果。在 `Choice` 和 `Wait` 状态下，`$` 指的是*有效输入*，即 `InputPath` 应用于状态输入之后的值。对于 `Pass`，`$` 指的是结果，即是由 `Result` 字段生成，还是由 `InputPath`/`Parameters` 字段生成。
下面的 JSONPath 示例将 JSON 对象分配给 `details` 变量，将 JSONPath 表达式 `$.result.code` 的结果分配给变量 `resultCode`，将 JSONPath 表达式 `States.Format('Hello {}', $customer.name)` 的结果分配给 `message`。如果处于 `Task` 状态，则 `$.order.items` 和 `$.result.code` 中的 `$` 指的是 API 结果。为 `startTime` 变量分配来自上下文对象 `$$.Execution.StartTime` 的值。
```
"Assign": {
"details": {
"status": "SUCCESS",
"lineItems.$": "$.order.items"
},
"resultCode.$": "$.result.code",
"message.$": "States.Format('Hello {}', $customer.name)",
"startTime.$": "$$.Execution.StartTime"
}
```
# 在 Step Functi JSONata ons 中使用转换数据
借 JSONata助，您可以获得强大的开源查询和表达式语言，用于在工作流程**中选择**和**转换**数据。有关简要介绍和完整 JSONata 参考资料，请参阅 [JSONata.org 文档](https://docs.jsonata.org/overview.html)。
**支持的 JSONata 版本**
Step Fun JSONata ctions 支持 2.0.6 版。
以下视频描述了变量，并 JSONata 在 Step Functions 中以 DynamoDB 为例：
您必须选择使用现有工作流程的 JSONata 查询和转换语言。在控制台中创建工作流程时，我们建议选择 JSONata 顶级状态机`QueryLanguage`。对于使用的现有工作流程或新工作流程 JSONPath，控制台提供了将单个状态转换为的选项 JSONata。
选择后 JSONata，您的工作流程字段将从五个 JSONPath 字段（`InputPath`、`Parameters`、`ResultSelector``ResultPath`、和`OutputPath`）减少到只有两个字段：`Arguments`和`Output`。此外，您**不会**在 JSON 对象键名称中使用 `.$`。
如果您不熟悉 Step Functions，则只需要知道 JSONata 表达式使用以下语法即可：
**JSONata 语法：**`"{% %}"`
以下代码示例显示了从 JSONPath 到的转换 JSONata：
```
# Original sample using JSONPath
{
"QueryLanguage": "JSONPath", // Set explicitly; could be set and inherited from top-level
"Type": "Task",
...
"Parameters": {
"static": "Hello",
"title.$": "$.title",
"name.$": "$customerName", // With $customerName declared as a variable
"not-evaluated": "$customerName"
}
}
```
```
# Sample after conversion to JSONata
{
"QueryLanguage": "JSONata", // Set explicitly; could be set and inherited from top-level
"Type": "Task",
...
"Arguments": { // JSONata states do not have Parameters
"static": "Hello",
"title": "{% $states.input.title %}",
"name": "{% $customerName %}", // With $customerName declared as a variable
"not-evaluated": "$customerName"
}
}
```
给定分配给 `"María"` 的输入 `{ "title" : "Doctor" }` 和变量 `customerName`，两台状态机都将生成以下 JSON 结果：
```
{
"static": "Hello",
"title": "Doctor",
"name": "María",
"not-evaluated": "$customerName"
}
```
在下图中，你可以看到一个图形表示形式，显示了转换 JSONPath （左）到 JSONata （右）将如何降低状态机中步骤的复杂性：
![\[比较 JSONPath 和 JSONata 状态中的字段的示意图。\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/compare-jsonpath-jsonata.png)
您可以（可选）从状态输入中选择数据，并将其转换为**参数**以发送到您的集成操作。然后 JSONata，您可以（可选）选择并转换操作的**结果**，以分配给变量和状态**输出**。
注意：**分配**和**输出**步骤**并行**进行。如果您选择在变量赋值期间转换数据，则转换后的数据将**不**可用于“输出”步骤。必须在 “输出” 步骤中重新应用 JSONata 转换。
![\[使用 JSONata 查询语言的状态的逻辑图。\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/vars-jsonata.png)
## QueryLanguage 字段
在您的工作流 ASL 定义中，有一个位于状态机定义顶级和各个状态中的 `QueryLanguage` 字段。通过`QueryLanguage`在单个状态内进行设置，可以在现有状态机 JSONata 中逐步采用，而不是一次性升级所有状态机。
`QueryLanguage` 字段可设置为 `"JSONPath"` 或 `"JSONata"`。如果省略顶级 `QueryLanguage` 字段，则默认为 `"JSONPath"`。如果状态包含状态级 `QueryLanguage` 字段，则 Step Functions 将使用该状态的指定查询语言。如果该状态不包含 `QueryLanguage` 字段，则它将使用顶级 `QueryLanguage` 字段中指定的查询语言。
## 用 JSON 字符串编写 JSONata 表达式
当 ASL 字段、JSON 对象字段或 JSON 数组元素的值中的字符串被`{% %}`字符包围时，该字符串的计算方式将为 JSONata 。注意，字符串必须以没有前导空格的 `{%` 开头，并且必须以没有尾随空格的 `%}` 结尾。表达式打开或关闭不当将会导致验证错误。
一些示例：
+ `"TimeoutSeconds" : "{% $timeout %}"`
+ `Task` 状态下的 `"Arguments" : {"field1" : "{% $name %}"}`
+ `Map` 状态下的 `"Items": [1, "{% $two %}", 3]`
并非所有 ASL 字段都接受 JSONata。例如，必须将每个状态的 `Type` 字段设置为常量字符串。同样，`Task` 状态的 `Resource` 字段必须是常量字符串。`Map`状态`Items`字段将接受 JSON 数组、JSON 对象或必须计算为数组或对象的 JSONata 表达式。
## 保留变量：\$1states
Step Functions 定义了一个名为的保留变量**`$states`**。在 JSONata 状态中，将以下结构分配`$states`给 JSONata 表达式中使用：
```
# Reserved $states variable in JSONata states
$states = {
"input": // Original input to the state
"result": // API or sub-workflow's result (if successful)
"errorOutput": // Error Output (only available in a Catch)
"context": // Context object
}
```
进入状态时，Step Functions 会将状态输入分配给。**`$states.input`**的值`$states.input`可用于所有接受 JSONata 表达式的字段。 `$states.input`总是指原始状态输入。
对于 `Task`、`Parallel` 和 `Map` 状态：
+ **`$states.result`**如果成功，则指 API 或子工作流程的原始结果。
+ **`$states.errorOutput`**指的是 API 或子工作流程失败时的错误输出。
`$states.errorOutput` 可用于 `Catch` 字段的 `Assign` 或 `Output`。
如果在无法访问 `$states.result` 或 `$states.errorOutput` 的字段和状态中尝试对其进行访问，则在创建、更新或验证状态机时将捕获该尝试。
`$states.context` 对象为您的工作流提供有关其具体执行的信息，例如 `StartTime`、任务令牌和初始工作流输入。要了解更多信息，请参阅[在 Step Functions 中从上下文对象访问执行数据](input-output-contextobject.md)。
## 处理表达式错误
在运行时， JSONata 表达式求值可能由于多种原因而失败，例如：
+ **类型错误** - 如果 `$x` 或 `$y` 不是数字，则表达式（例如 `{% $x + $y %}`）将失败。
+ **类型不兼容** - 表达式的计算结果可能为该字段不接受的类型。例如，字段 `TimeoutSeconds` 需要数字输入，因此如果 `$timeout` 返回字符串，则表达式 `{% $timeout %}` 将失败。
+ **值超出范围** - 生成超出字段可接受范围的值的表达式将失败。例如，诸如 `{% $evaluatesToNegativeNumber %}` 之类的表达式在 `TimeoutSeconds` 字段中将会失败。
+ **未能返回结果** - JSON 不能表示未定义的值表达式，因此表达式 `{% $data.thisFieldDoesNotExist %}` 会导致错误。
在每种情况下，解释器都会引发错误：`States.QueryEvaluationError`。您的 Task、Map 和 Parallel 状态可以提供一个 `Catch` 字段用于捕获错误，并提供一个 `Retry` 字段用于在出错时重试。
## 从 JSONPath 转换为 JSONata
以下各节比较并解释了使用 JSONPath 和编写的代码之间的区别 JSONata。
### 没有更多路径字段
ASL 要求开发人员在使用时使用 JSONPath字段`Path`版本从状态数据中选择值，如中所`TimeoutSecondsPath`示。使用时 JSONata，您将不再使用`Path`字段，因为 ASL 会自动为您解释非路径字段中`{% %}`包含的 JSONata 表达式，例如。`TimeoutSeconds`
+ JSONPath 旧版示例：`"TimeoutSecondsPath": "$timeout"`
+ JSONata : `"TimeoutSeconds": "{% $timeout %}"`
同样，`Map`状态`ItemsPath`已替换为接受 JSON 数组、JSON 对象或必须计算为数组或对象的 JSONata 表达式的`Items`字段。
### JSON 对象
ASL 使用术语*有效载荷模板*来描述可以包含`Parameters`和`ResultSelector`字段值 JSONPath 表达式的 JSON 对象。ASL 不会使用术语有效负载模板， JSONata 因为所有字符串都会 JSONata 进行评估，无论它们是单独出现的，还是出现在 JSON 对象或 JSON 数组中。
### 没有更多 .\$1
ASL 要求您在要使用的有效载荷模板 JSONPath 和内部函数中的字段名称后附加 `.$` “”。指定 `"QueryLanguage":"JSONata"` 时，您不再对 JSON 对象字段名称使用“`.$`”约定，而是用`{% %}`字符将 JSONata 表达式括起来。无论对象嵌套在其他数组或对象中的深度如何，您都对所有字符串值字段使用相同的约定。
### Arguments 和 Output 字段
当设置`QueryLanguage`为时`JSONata`，旧的 I/O 处理字段将被禁用（`InputPath`、`Parameters``ResultSelector`、`ResultPath`和`OutputPath`），并且大多数州将获得两个新字段：`Arguments`和`Output`。
JSONata 与使用的字段相比，提供了一种更简单的 I/O 转换方法。 JSONPath JSONata的功能比前五个字段`Output`更强大 JSONPath。`Arguments`这些新的字段名称还有助于简化 ASL，并阐明值传递和返回模型。
`Arguments` 和 `Output` 字段（以及其他类似字段，如 `Map` 状态的 `ItemSelector`）将接受 JSON 对象，例如：
```
"Arguments": {
"field1": 42,
"field2": "{% jsonata expression %}"
}
```
或者，你可以直接使用 JSONata 表达式，例如：
```
"Output": "{% jsonata expression %}"
```
输出也可以接受任何类型的 JSON 值，例如：`"Output":true`、`"Output":42`。
`Arguments`和`Output`字段仅支持 JSONata，因此在使用的工作流程中使用它们是无效的 JSONPath。相反，`InputPath`、`Parameters`、`ResultSelector`、`ResultPath``OutputPath`、和其他 JSONPath 字段仅在中受支持 JSONPath，因此在 JSONata 用作顶级工作流程或状态查询语言时，使用基于路径的字段是无效的。
### Pass 状态
Pass 状态中的可选**结果**以前被视为虚拟任务的*输出*。 JSONata 选择为工作流或状态查询语言后，您现在可以使用新的**输出**字段。
### Choice 状态
使用时 JSONPath，选择状态具有输入`Variable`和许多比较路径，如下所示`NumericLessThanEqualsPath`：
```
# JSONPath choice state sample, with Variable and comparison path
"Check Price": {
"Type": "Choice",
"Default": "Pause",
"Choices": [
{
"Variable": "$.current_price.current_price",
"NumericLessThanEqualsPath": "$.desired_price",
"Next": "Send Notification"
} ],
}
```
使用 JSONata，选择状态有一个你可以使用 JSONata 表达式`Condition`的地方：
```
# Choice state after JSONata conversion
"Check Price": {
"Type": "Choice",
"Default": "Pause"
"Choices": [
{
"Condition": "{% $current_price <= $states.input.desired_priced %}",
"Next": "Send Notification"
} ]
```
注意：变量和比较字段仅适用于 JSONPath。状况仅适用于 JSONata。
## JSONata 例子
可以在 Workflow Studio 中创建以下示例进行实验 JSONata。您可以创建和执行状态机，也可以使用 **Test 状态**来传入数据，甚至修改状态机定义。
### 示例：输入和输出
此示例说明在您选择加入时`$states.input`如何使用状态输入和`Output`字段来指定状态输出 JSONata。
```
{
"Comment": "Input and Output example using JSONata",
"QueryLanguage": "JSONata",
"StartAt": "Basic Input and Output",
"States": {
"Basic Input and Output": {
"QueryLanguage": "JSONata",
"Type": "Succeed",
"Output": {
"lastName": "{% 'Last=>' & $states.input.customer.lastName %}",
"orderValue": "{% $states.input.order.total %}"
}
}
}
}
```
当使用以下内容作为输入来执行工作流时：
```
{
"customer": {
"firstName": "Martha",
"lastName": "Rivera"
},
"order": {
"items": 7,
"total": 27.91
}
}
```
Test 状态或状态机执行将返回以下 JSON 输出：
```
{
"lastName": "Last=>Rivera",
"orderValue": 27.91
}
```
![\[屏幕截图显示了被测状态的输入和输出。\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/jsonata-basic-io.png)
### 示例：使用筛选 JSONata
您可以使用 JSONata [路径运算符](https://docs.jsonata.org/path-operators)筛选数据。例如，假设您有一份可供输入的产品清单，而您只想处理卡路里含量为零的产品。您可以使用以下 ASL 创建状态机定义，并使用随后的示例输入来测试 `FilterDietProducts` 状态。
**用于筛选的状态机定义 JSONata**
```
{
"Comment": "Filter products using JSONata",
"QueryLanguage": "JSONata",
"StartAt": "FilterDietProducts",
"States": {
"FilterDietProducts": {
"Type": "Pass",
"Output": {
"dietProducts": "{% $states.input.products[calories=0] %}"
},
"End": true
}
}
}
```
**测试的示例输入**
```
{
"products": [
{
"calories": 140,
"flavour": "Cola",
"name": "Product-1"
},
{
"calories": 0,
"flavour": "Cola",
"name": "Product-2"
},
{
"calories": 160,
"flavour": "Orange",
"name": "Product-3"
},
{
"calories": 100,
"flavour": "Orange",
"name": "Product-4"
},
{
"calories": 0,
"flavour": "Lime",
"name": "Product-5"
}
]
}
```
**状态机中步骤测试的输出**
```
{
"dietProducts": [
{
"calories": 0,
"flavour": "Cola",
"name": "Product-2"
},
{
"calories": 0,
"flavour": "Lime",
"name": "Product-5"
}
]
}
```
![\[被测 JSONata 表达式的输出示例。\]](http://docs.aws.amazon.com/zh_cn/step-functions/latest/dg/images/test-state-jsonata.png)
## JSONata Step Functions 提供的函数
JSONata 包含字符串、数字、聚合、布尔函数、数组、对象、日期/时间和高阶函数的函数库。Step Functions 提供了可以在 JSONata 表达式中使用的其他 JSONata 函数。这些内置函数可以替换 Step Functions 内置函数。内部函数仅在使用 JSONPath 查询语言的状态下可用。
注意：需要整数值作为参数的内置 JSONata 函数会自动向下舍入所提供的任何非整数。
**\$1partiti JSONata on-** 等同于对大型数组进行分区的`States.ArrayPartition`内在函数。
第一个参数是要分区的数组，第二个参数是代表区块大小的整数。返回值将是一个二维数组。解释器将输入数组分成多个数组，其大小由区块大小指定。如果数组中剩余的项目数小于区块大小，则最后一个数组区块的长度可能小于之前的数组区块的长度。
```
"Assign": {
"arrayPartition": "{% $partition([1,2,3,4], $states.input.chunkSize) %}"
}
```
**\$1rang** e- JSONata 相当于生成值数组的`States.ArrayRange`内在函数。
这个函数需要三个参数。第一个参数是代表新数组第一个元素的整数，第二个参数是代表新数组最后一个元素的整数，第三个参数是新数组中元素的增量值整数。返回值是一个新生成的值数组，范围从函数的第一个参数到函数的第二个参数，两者之间的元素根据增量进行调整。增量值可以是正值，也可以是负值，它将从最后一个元素开始依次递增或递减，直至达到或超过最终值。
```
"Assign": {
"arrayRange": "{% $range(0, 10, 2) %}"
}
```
**\$1hash**- JSONata 等效于计算给定`States.Hash`输入哈希值的内在函数。
这个函数需要两个参数。第一个参数是要进行哈希处理的源字符串。第二个参数是代表用于哈希计算的哈希算法的字符串。哈希算法必须是以下值之一：`"MD5"`、`"SHA-1"`、`"SHA-256"`、`"SHA-384"`、`"SHA-512"`。返回值是计算得出的数据哈希值的字符串。
之所以创建此函数，是因为 JSONata 它本身不支持计算哈希的功能。
```
"Assign": {
"myHash": "{% $hash($states.input.content, $hashAlgorithmName) %}"
}
```
**\$1random- JSONata 等同于返回随**机数 n 的`States.MathRandom`内在函数，其中。`0 ≤ n < 1`
该函数接受一个*可选*的整数参数，表示随机函数的种子值。如果您使用具有相同种子值的函数，它将返回相同的数字。
之所以创建这个重载函数，是因为内置 JSONata 函数[https://docs.jsonata.org/numeric-functions#random](https://docs.jsonata.org/numeric-functions#random)不接受种子值。
```
"Assign": {
"randNoSeed": "{% $random() %}",
"randSeeded": "{% $random($states.input.seed) %}"
}
```
**\$1uuid**-`States.UUID` 内部函数的 JSONata 版本。
此函数不接受任何参数。此函数会返回一个 v4 UUID。
之所以创建此函数，是因为 JSONata 它本身不支持生成 UUIDs功能。
```
"Assign": {
"uniqueId": "{% $uuid() %}"
}
```
**\$1parse**-用于反序列化 JSON 字符串的 JSONata 函数。
此函数将字符串化的 JSON 作为其唯一参数。
JSONata 通过支持此功能`$eval`；但是，Step Fun `$eval` ctions 工作流程不支持此功能。
```
"Assign": {
"deserializedPayload": "{% $parse($states.input.json_string) %}"
}
```
# 在 Step Functions 中从上下文对象访问执行数据
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
上下文对象是执行期间可用的内部 JSON 结构，包含有关状态机和执行的信息。上下文可为您的工作流提供有关其特定执行的信息。您的工作流程可以使用在 JSONata 表达式中引用 Context 对象`$states.context`。
## 访问上下文对象
**要访问中的上下文对象 JSONata**
要在 JSONata 状态下访问 Context 对象，请在 JSONata 表达式`$states.context`中使用。
```
{
"ExecutionID" : "{% $states.context.Execution.Id %}"
}
```
**要访问中的上下文对象 JSONPath**
要访问中的 Context 对象 JSONPath，首先`.$`要在键的末尾追加以指示该值是路径。然后，在值前加上 `$$.` 以选择上下文对象中的节点。
```
{
"ExecutionID.$": "$$.Execution.Id"
}
```
JSONPath 状态可以从以下 JSONPath字段中引用上下文 (`$$.`)：
+ `InputPath`
+ `OutputPath`
+ `ItemsPath`（在 Map 状态下）
+ `Variable`（在 Choice 状态下）
+ `ResultSelector`
+ `Parameters`
+ 变量到变量的比较运算符
## 上下文对象字段
上下文对象包括有关状态机、状态、执行和任务的信息。该上下文 JSON 对象包括每种类型数据的节点，并采用以下格式：
```
{
"Execution": {
"Id": "String",
"Input": {},
"Name": "String",
"RoleArn": "String",
"StartTime": "Format: ISO 8601",
"RedriveCount": Number,
"RedriveTime": "Format: ISO 8601"
},
"State": {
"EnteredTime": "Format: ISO 8601",
"Name": "String",
"RetryCount": Number
},
"StateMachine": {
"Id": "String",
"Name": "String"
},
"Task": {
"Token": "String"
}
}
```
在执行期间，上下文对象将以相关数据填充。
有时，上下文中会添加新字段。如果您要直接处理 JSON 上下文，建议您编写能够妥善处理新的未知字段的代码。例如，如果使用 Jackson 库对 JSON 进行解组，建议在您的 `ObjectMapper` 中将 `FAIL_ON_UNKNOWN_PROPERTIES` 设置为 `false` 以防止 `UnrecognizedPropertyException`。
`RedriveTime` 上下文对象只有在您已redriven执行时才可用。如果您已[redriven a Map Run](redrive-map-run.md)，则 `RedriveTime` 上下文对象仅适用于标准类型的子工作流。对于使用快速类型的子工作流的redriven Map Run，`RedriveTime` 不可用。
来自正在运行的执行的内容包括以下格式的具体信息：
```
{
"Execution": {
"Id": "arn:aws:states:region:123456789012:execution:stateMachineName:executionName",
"Input": {
"key": "value"
},
"Name": "executionName",
"RoleArn": "arn:aws:iam::123456789012:role...",
"StartTime": "2025-08-27T10:04:42Z"
},
"State": {
"EnteredTime": "2025-08-27T10:04:42.001Z",
"Name": "Test",
"RetryCount": 3
},
"StateMachine": {
"Id": "arn:aws:states:region:123456789012:stateMachine:stateMachineName",
"Name": "stateMachineName"
},
"Task": {
"Token": "h7XRiCdLtd/83p1E0dMccoxlzFhglsdkzpK9mBVKZsp7d9yrT1W"
}
}
```
**具有小数秒精度的时间戳格式**
Step Functions 遵循 ISO8601 规范，即根据需要，输出可以是零、三、六或九位数字。当时间戳的小数秒部分为零时，Step Functions 会移除末尾的零，而不会对输出进行补零填充。
如果您创建的代码使用 Step Functions 时间戳，则您的代码必须能够处理可变数量的小数秒。
## Map 状态的上下文对象数据
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
处理 [`Map` 状态](state-map.md)时，上下文还将包含 `Index`、`Value` 和 `Source`。
对于每个`Map`状态迭代，`Index`包含当前正在处理的数组项的索引号，`Value`包含正在处理的数组项，`Source`并将是`CSV``JSON`、`JSONL`、或`PARQUET`。 InputType
在 `Map` 状态内，上下文对象包括以下数据：
```
"Map": {
"Item": {
"Index" : Number,
"Key" : "String", // Only valid for JSON objects
"Value" : "String",
"Source": "String"
}
}
```
这些项仅在 `Map` 状态下可用，并且可以在 `ItemSelector（Map）` 字段中指定。
**注意**
您必须在主要 `ItemSelector` 状态的 `Map` 块中的上下文对象中定义参数，而不是在 `ItemProcessor` 部分中包含的状态中定义参数。
给定一个使用状态的**JSONPath**`Map`状态机，你可以按如下方式从 Context 对象中注入信息。
```
{
"StartAt": "ExampleMapState",
"States": {
"ExampleMapState": {
"Type": "Map",
"ItemSelector": {
"ContextIndex.$": "$$.Map.Item.Index",
"ContextValue.$": "$$.Map.Item.Value",
"ContextSource.$": "$$.Map.Item.Source"
},
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "INLINE"
},
"StartAt": "TestPass",
"States": {
"TestPass": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
```
对于 JSONata，可以从`$states.context`变量访问其他 Map 状态上下文信息：
```
{
"StartAt": "ExampleMapState",
"States": {
"ExampleMapState": {
"Type": "Map",
"ItemSelector": {
"ContextIndex": "{% $states.context.Map.Item.Index %}",
"ContextValue": "{% $states.context.Map.Item.Value %}",
"ContextSource": "{% $states.context.Map.Item.Source %}"
},
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "INLINE"
},
"StartAt": "TestPass",
"States": {
"TestPass": {
"Type": "Pass",
"End": true
}
}
},
"End": true
}
}
}
```
如果使用以下输入执行前一状态机，则将 `Index` 和 `Value` 插入到输出中。
```
[
{
"who": "bob"
},
{
"who": "meg"
},
{
"who": "joe"
}
]
```
执行的输出返回三次迭代中每次迭代的 `Index` 和 `Value` 项的值，如下所示：
```
[
{
"ContextIndex": 0,
"ContextValue": {
"who": "bob"
},
"ContextSource" : "STATE_DATA"
},
{
"ContextIndex": 1,
"ContextValue": {
"who": "meg"
},
"ContextSource" : "STATE_DATA"
},
{
"ContextIndex": 2,
"ContextValue": {
"who": "joe"
},
"ContextSource" : "STATE_DATA"
}
]
```
请注意，`$states.context.Map.Item.Source` 将是以下项之一：
+ 对于状态输入，该值将为：`STATE_DATA`
+ 对于 `Amazon S3 LIST_OBJECTS_V2` (`Transformation=NONE`)，该值将显示存储桶的 S3 URI。例如：`S3://bucket-name`。
+ 对于所有其他输入类型，该值将为 Amazon S3 URI。例如：`S3://bucket-name/object-key`。
# 使用 JSONPath 路径
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
在 Amazon States Language中，*路径*是以 `$` 开头的字符串，您可使用该字符串来标识 JSON 文本中的组件。路径遵循[JsonPath](https://datatracker.ietf.org/wg/jsonpath/about/)语法，该语法仅在设置`QueryLanguage`为时才可用 JSONPath。在指定 `InputPath`、`ResultPath` 和 `OutputPath` 的值时，可以指定访问输入子集的路径。
如果您的字段名包含任何未包含在 [JsonPath ABNF](https://www.ietf.org/archive/id/draft-ietf-jsonpath-base-21.html#jsonpath-abnf) 规则`member-name-shorthand`定义中的字符，则必须使用方括号表示法。因此，要对标点符号（不包括`_`）等特殊字符进行编码，必须使用方括号表示法。例如 `$.abc.['def ghi']`。
## 引用路径
*引用路径* 是一种语法存在限制的路径，它只能标识 JSON 结构中的单个节点：
+ 您可以使用句点 (`.`) 和方括号 (`[ ]`) 表示法访问对象字段。
+ 不支持像 `length()` 这样的函数。
+ 不支持非符号的词法运算符，例如 `subsetof`。
+ 不支持按正则表达式或通过引用 JSON 结构中的其它值来进行筛选。
+ 不支持 `@`、`,`、`:` 和 `?` 运算符。
例如，如果状态输入数据包含以下值：
```
{
"foo": 123,
"bar": ["a", "b", "c"],
"car": {
"cdr": true
}
}
```
以下引用路径将返回以下内容。
```
$.foo => 123
$.bar => ["a", "b", "c"]
$.car.cdr => true
```
某些状态使用路径和引用路径来控制状态机的流，或者配置状态的设置或选项。有关更多信息，请参阅使用[数据流模拟器对工作流输入和输出路径处理](https://aws.amazon.com/blogs/compute/modeling-workflow-input-output-path-processing-with-data-flow-simulator/)进行建模和[ JSONPath 有效使用AWS Step Functions](https://aws.amazon.com/blogs/compute/using-jsonpath-effectively-in-aws-step-functions/)。
### 展平由数组组成的数组
如果状态机中的 [Parallel 工作流程状态](state-parallel.md) 或 [Map 状态工作流程。](state-map.md) 状态返回由数组组成的数组，则可以使用 [ResultSelector](input-output-inputpath-params.md#input-output-resultselector) 字段将他们转换为一个平面数组。您可以将此字段包含在 Parallel 或 Map 状态定义中，以操纵这些状态的结果。
要展平数组，请在 `ResultSelector` 字段中使用语法：`[*]`，如以下示例所示。
```
"ResultSelector": {
"flattenArray.$": "$[*][*]"
}
```
有关演示如何展平数组的示例，请参阅以下教程中的*第 3 步*：
+ [在 Step Functions 中使用 Lambda 函数处理批量数据](tutorial-itembatcher-param-task.md)
+ [在 Step Functions 中使用 Lambda 函数处理各个项目](tutorial-itembatcher-single-item-process.md)
# 在 Step Functions 工作流中操作参数
**管理状态和转换数据**
了解有关[使用变量在状态之间传递数据](workflow-variables.md)和[使用转换数据](transforming-data.md)的信息 JSONata。
`InputPath`、`Parameters` 和 `ResultSelector` 字段提供了一种在 JSON 流经工作流时操纵它的方法。`InputPath` 可以通过使用路径筛选 JSON 表示法来限制传递的输入（请参阅[使用 JSONPath 路径](amazon-states-language-paths.md)）。借助 `Parameters` 字段，您可以使用静态值或通过路径输入的选择来传递一组键值对。
`ResultSelector` 字段提供了一种在应用 `ResultPath` 状态之前操作状态结果的方法。
AWS Step Functions首先应用`InputPath`字段，然后应用字`Parameters`段。您可以先使用 `InputPath` 将原始输入筛选为所选内容，然后应用 `Parameters` 进一步操作该输入，或添加新值。然后，您可以在应用 `ResultPath` 状态之前使用 `ResultSelector` 字段来操作状态的输出。
## InputPath
使用 `InputPath` 选择状态输入的一部分。
例如，假设状态输入包括以下内容。
```
{
"comment": "Example for InputPath.",
"dataset1": {
"val1": 1,
"val2": 2,
"val3": 3
},
"dataset2": {
"val1": "a",
"val2": "b",
"val3": "c"
}
}
```
您可以应用 `InputPath`。
```
"InputPath": "$.dataset2",
```
借助之前的 `InputPath`，以下内容是作为输入传递的 JSON。
```
{
"val1": "a",
"val2": "b",
"val3": "c"
}
```
**注意**
一条路径可以生成一系列值。考虑以下示例。
```
{ "a": [1, 2, 3, 4] }
```
如果您应用路径 `$.a[0:2]`，结果如下。
```
[ 1, 2 ]
```
## Parameters
本部分介绍您可以使用 Parameters 字段的不同方式。
### 键值对
使用 `Parameters` 字段创建作为输入传递的键值对集合。每个键值对的值可以是您在状态机定义中包含的静态值，也可以是使用路径从输入或上下文对象中选择的值。对于使用路径选择值的键值对，键名必须以 `.$` 结尾。
例如，假设您提供以下输入。
```
{
"comment": "Example for Parameters.",
"product": {
"details": {
"color": "blue",
"size": "small",
"material": "cotton"
},
"availability": "in stock",
"sku": "2317",
"cost": "$23"
}
}
```
要选择一些信息，您可以在状态机定义中指定这些参数。
```
"Parameters": {
"comment": "Selecting what I care about.",
"MyDetails": {
"size.$": "$.product.details.size",
"exists.$": "$.product.availability",
"StaticValue": "foo"
}
},
```
给定上一个输入和 `Parameters` 字段，这是传递的 JSON。
```
{
"comment": "Selecting what I care about.",
"MyDetails": {
"size": "small",
"exists": "in stock",
"StaticValue": "foo"
}
},
```
除了输入之外，您还可以访问一个特殊的 JSON 对象，称为上下文对象。上下文对象包含有关状态机执行的信息。请参阅[在 Step Functions 中从上下文对象访问执行数据](input-output-contextobject.md)。
### 连接的资源
`Parameters` 字段还可以将信息传递给连接的资源。例如，如果您的任务状态为编排AWS Batch任务，则可以将相关的 API 参数直接传递给该服务的 API 操作。有关更多信息，请参阅:
+ [在 Step Functions 中将参数传递给服务 API](connect-parameters.md)
+ [集成 服务](integrate-services.md)
### Amazon S3
如果您在各状态之间传递的 Lambda 函数数据可能增长到超过 262,144 字节，我们建议使用 Amazon S3 来存储数据，并实施以下方法之一：
+ 在工作流中使用*分布式 Map 状态*，以便 `Map` 状态可以直接从 Amazon S3 数据来源读取输入。有关更多信息，请参阅 [分布式模式](state-map-distributed.md)。
+ 解析 `Payload` 参数中存储桶的 Amazon 资源名称 (ARN)，以获取存储桶名称和键值。有关更多信息，请参阅 [使用 Amazon S3 ARNs 而不是在 Step Functions 中传递大型有效负载](sfn-best-practices.md#avoid-exec-failures)。
或者，您也可以调整实施，以便在执行中传递较少的有效负载。
## ResultSelector
在应用 `ResultPath` 之前，使用 `ResultSelector` 字段来操作状态的结果。`ResultSelector` 字段可用于创建键值对的集合，其中值为静态值或从状态的结果中选择。使用 `ResultSelector` 字段，您可以选择要将状态结果的哪些部分传递给 `ResultPath` 字段。
**注意**
使用 `ResultPath` 字段，您可以将 `ResultSelector` 字段的输出添加到原始输入中。
`ResultSelector` 是处于以下状态的可选字段：
+ [Map 状态工作流程。](state-map.md)
+ [Task 工作流程状态](state-task.md)
+ [Parallel 工作流程状态](state-parallel.md)
例如，除了结果中的有效负载外，Step Functions 服务集成还会返回元数据。`ResultSelector` 可以选择结果的某些部分并使用 `ResultPath` 将其与状态输入合并。在此示例中，我们只想选择 `resourceType` 和 `ClusterId`，然后将其与来自 Amazon EMR createCluster.sync 的状态输入合并。给定以下内容：
```
{
"resourceType": "elasticmapreduce",
"resource": "createCluster.sync",
"output": {
"SdkHttpMetadata": {
"HttpHeaders": {
"Content-Length": "1112",
"Content-Type": "application/x-amz-JSON-1.1",
"Date": "Mon, 25 Nov 2019 19:41:29 GMT",
"x-amzn-RequestId": "1234-5678-9012"
},
"HttpStatusCode": 200
},
"SdkResponseMetadata": {
"RequestId": "1234-5678-9012"
},
"ClusterId": "AKIAIOSFODNN7EXAMPLE"
}
}
```
然后，您可以使用 `ResultSelector` 选择 `resourceType` 和 `ClusterId`：
```
"Create Cluster": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:createCluster.sync",
"Parameters": {
},
"ResultSelector": {
"ClusterId.$": "$.output.ClusterId",
"ResourceType.$": "$.resourceType"
},
"ResultPath": "$.EMROutput",
"Next": "Next Step"
}
```
使用给定的输入，使用 `ResultSelector` 会产生：
```
{
"OtherDataFromInput": {},
"EMROutput": {
"ClusterId": "AKIAIOSFODNN7EXAMPLE",
"ResourceType": "elasticmapreduce",
}
}
```
### 展平由数组组成的数组
如果状态机中的 [Parallel 工作流程状态](state-parallel.md) 或 [Map 状态工作流程。](state-map.md) 状态返回由数组组成的数组，则可以使用 [ResultSelector](#input-output-resultselector) 字段将他们转换为一个平面数组。您可以将此字段包含在 Parallel 或 Map 状态定义中，以操纵这些状态的结果。
要展平数组，请在 `ResultSelector` 字段中使用语法：`[*]`，如以下示例所示。
```
"ResultSelector": {
"flattenArray.$": "$[*][*]"
}
```
有关演示如何展平数组的示例，请参阅以下教程中的*第 3 步*：
+ [在 Step Functions 中使用 Lambda 函数处理批量数据](tutorial-itembatcher-param-task.md)
+ [在 Step Functions 中使用 Lambda 函数处理各个项目](tutorial-itembatcher-single-item-process.md)
# 示例：在 Step Functions 工作流中操作带路径的状态数据
**管理状态和转换数据**
了解如何[使用变量在状态之间传递数据](workflow-variables.md)和[使用 JSONata 转换数据](transforming-data.md)。
本主题包含有关如何使用 InputPath、ResultPath 和 OutputPath 字段来操作状态输入和输出 JSON 的示例。
除 [Fail 工作流程状态](state-fail.md) 或 [Succeed 工作流程状态](state-succeed.md) 状态外的任何状态，都可以包含输入和输出处理字段，如 `InputPath`、`ResultPath` 或 `OutputPath`。此外，[Wait 工作流程状态](state-wait.md) 和 [Choice 工作流程状态](state-choice.md) 状态不支持 `ResultPath` 字段。通过这些字段，您可以使用 [JsonPath](https://datatracker.ietf.org/wg/jsonpath/about/) 来筛选在工作流中移动的 JSON 数据。
您还可以使用 `Parameters` 字段在 JSON 数据在工作流中移动时对其进行操作。有关使用 `Parameters` 的信息，请参阅 [在 Step Functions 工作流中操作参数](input-output-inputpath-params.md)。
例如，从AWS Lambda教程中所述的 [创建使用 Lambda 的 Step Functions 状态机](tutorial-creating-lambda-state-machine.md) 函数和状态机开始。修改状态机，以便它包含以下 `InputPath`、`ResultPath` 和 `OutputPath`。
```
{
"Comment": "A Hello World example of the Amazon States Language using an AWS Lambda function",
"StartAt": "HelloWorld",
"States": {
"HelloWorld": {
"Type": "Task",
"Resource": "arn:aws:lambda:region:123456789012:function:HelloFunction",
"InputPath": "$.lambda",
"ResultPath": "$.data.lambdaresult",
"OutputPath": "$.data",
"End": true
}
}
}
```
使用以下输入开始执行。
```
{
"comment": "An input comment.",
"data": {
"val1": 23,
"val2": 17
},
"extra": "foo",
"lambda": {
"who": "AWS Step Functions"
}
}
```
假设可以丢弃 `comment` 和 `extra` 节点，但您想要包含 Lambda 函数的输出以及保留 `data` 节点中的信息。
在更新后的状态机中，`Task` 状态将更改以处理任务输入。
```
"InputPath": "$.lambda",
```
状态机定义中的此行将任务输入限制为仅状态输入中的 `lambda` 节点。Lambda 函数仅接收输入形式的 JSON 对象 `{"who": "AWS Step Functions"}`。
```
"ResultPath": "$.data.lambdaresult",
```
此 `ResultPath` 指示状态机将 Lambda 函数的结果插入名为 `lambdaresult`、作为原始状态机输入中 `data` 节点的子级的节点中。因为您没有使用 `OutputPath` 对原始输入和结果执行任何其它操作，所以现在状态的输出包括 Lambda 函数的结果以及原始输入。
```
{
"comment": "An input comment.",
"data": {
"val1": 23,
"val2": 17,
"lambdaresult": "Hello, AWS Step Functions!"
},
"extra": "foo",
"lambda": {
"who": "AWS Step Functions"
}
}
```
但是，我们的目标是仅保留 `data` 节点，并且包含 Lambda 函数的结果。`OutputPath` 将先筛选此组合的 JSON，然后再将其传递到状态输出。
```
"OutputPath": "$.data",
```
这将仅选择原始输入中要传递到输出的 `data` 节点（包括 `lambdaresult` 插入的 `ResultPath` 子节点）。状态输出将筛选为以下内容。
```
{
"val1": 23,
"val2": 17,
"lambdaresult": "Hello, AWS Step Functions!"
}
```
在该 `Task` 状态中：
1. `InputPath` 仅将输入中的 `lambda` 节点发送至 Lambda 函数。
1. `ResultPath` 将结果作为 `data` 节点的子级插入原始输入。
1. `OutputPath` 将筛选状态输入（现在包含 Lambda 函数的结果），以便它仅将 `data` 节点传递到状态输出。
**Example 使用 JsonPath 操作原始状态机输入、结果和最终输出**
考虑以下状态机器，它可以验证保险申请人的身份和地址。
要查看完整的示例，请参阅 [How to use JSON Path in Step Functions](https://github.com/aws-samples/serverless-account-signup-service)。
```
{
"Comment": "Sample state machine to verify an applicant's ID and address",
"StartAt": "Verify info",
"States": {
"Verify info": {
"Type": "Parallel",
"End": true,
"Branches": [
{
"StartAt": "Verify identity",
"States": {
"Verify identity": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"Parameters": {
"Payload.$": "$",
"FunctionName": "arn:aws:lambda:us-east-2:111122223333:function:check-identity:$LATEST"
},
"End": true
}
}
},
{
"StartAt": "Verify address",
"States": {
"Verify address": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"Parameters": {
"Payload.$": "$",
"FunctionName": "arn:aws:lambda:us-east-2:111122223333:function:check-address:$LATEST"
},
"End": true
}
}
}
]
}
}
}
```
如果您使用以下输入运行此状态机，则执行将失败，因为执行验证的 Lambda 函数只期望需要验证的数据作为输入。因此，您必须使用适当的 JsonPath 指定包含待验证信息的节点。
```
{
"data": {
"firstname": "Jane",
"lastname": "Doe",
"identity": {
"email": "jdoe@example.com",
"ssn": "123-45-6789"
},
"address": {
"street": "123 Main St",
"city": "Columbus",
"state": "OH",
"zip": "43219"
},
"interests": [
{
"category": "home",
"type": "own",
"yearBuilt": 2004
},
{
"category": "boat",
"type": "snowmobile",
"yearBuilt": 2020
},
{
"category": "auto",
"type": "RV",
"yearBuilt": 2015
},
]
}
}
```
要指定 `check-identity` Lambda 函数必须使用的节点，请使用 `InputPath` 字段，如下所示：
```
"InputPath": "$.data.identity"
```
要指定 `check-address` Lambda 函数必须使用的节点，请使用 `InputPath` 字段，如下所示：
```
"InputPath": "$.data.address"
```
现在，如果要将验证结果存储在原始状态机输入中，请使用 `ResultPath` 字段，如下所示：
```
"ResultPath": "$.results"
```
但是，如果您只需要身份和验证结果并丢弃原始输入，请使用 `OutputPath` 字段，如下所示：
```
"OutputPath": "$.results"
```
有关更多信息，请参阅 [在 Step Functions 中处理输入和输出](concepts-input-output-filtering.md)。
## 使用 OutputPath 筛选状态输出
使用 `OutputPath`，可以选择状态输出的一部分以传递到下一个状态。通过此方法，可以筛选掉不需要的信息，而仅传递您需要的这部分 JSON。
如果您未指定 `OutputPath`，则默认值是 `$`。这会将整个 JSON 节点（由状态输入、任务结果和 `ResultPath` 确定）传递到下一个状态。
# 在 Step Functions ResultPath 中使用指定状态输出
**管理状态和转换数据**
本页指的是 JSONPath。Step Functions 最近添加了变量 JSONata ，用于管理状态和转换数据。
了解如何[使用变量传递数据](workflow-variables.md)和[使用转换数据 JSONata](transforming-data.md)。
状态的输出可以是其输入的副本、其生成的结果（例如，`Task` 状态的 Lambda 函数的输出）或其输入和结果的组合。使用 `ResultPath` 可控制传递到状态输出的上述两种内容的组合。
以下状态类型可以生成结果且包含 `ResultPath:`
+ [Pass 工作流程状态](state-pass.md)
+ [Task 工作流程状态](state-task.md)
+ [Parallel 工作流程状态](state-parallel.md)
+ [Map 状态工作流程。](state-map.md)
使用 `ResultPath` 可将任务结果与任务输入组合，或可选择二者之一。提供给 `ResultPath` 的路径控制将传递到输出的信息。
**注意**
`ResultPath` 仅限于使用[引用路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)，这会限制其范围，因此该路径必须仅标识 JSON 中的单一节点。请参阅 [Amazon States Language](concepts-amazon-states-language.md) 中的 [引用路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)。
## 用于 ResultPath 用任务结果替换输入
如果未指定 `ResultPath`，则默认行为与 `"ResultPath": "$"` 相同。该状态会将整个状态输入替换为任务结果。
```
# State Input
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions"
}
# Path
"ResultPath": "$"
# Task result
"Hello, Step Functions!"
# State Output
"Hello, Step Functions!"
```
**注意**
`ResultPath` 用于包含结果内容与输入，然后再将其传递到输出。但是，如果未指定 `ResultPath`，则默认操作为替换整个输入。
## 丢弃结果并保留原始输入
如果将 `ResultPath` 设置为 `null`，则该状态会将**原始输入**传递给输出。该状态的输入有效载荷将直接复制到输出中，而不考虑任务结果。
```
# State Input
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions"
}
# Path
"ResultPath": null
# Task result
"Hello, Step Functions!"
# State Output
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions"
}
```
## 用于 ResultPath 将结果包含在输入中
如果您为指定路径 ResultPath，则状态输出将合并状态输入和任务结果：
```
# State Input
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions"
}
# Path
"ResultPath": "$.taskresult"
# Task result
"Hello, Step Functions!"
# State Output
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions",
"taskresult" : "Hello, Step Functions!"
}
```
还可以将结果插入输入的子节点。将 `ResultPath` 设置为以下内容。
```
"ResultPath": "$.strings.lambdaresult"
```
如果使用以下输入：
```
{
"comment": "An input comment.",
"strings": {
"string1": "foo",
"string2": "bar",
"string3": "baz"
},
"who": "AWS Step Functions"
}
```
任务结果将作为输入中 `strings` 节点的子节点插入。
```
{
"comment": "An input comment.",
"strings": {
"string1": "foo",
"string2": "bar",
"string3": "baz",
"lambdaresult": "Hello, Step Functions!"
},
"who": "AWS Step Functions"
}
```
状态输出现在包含原始输入 JSON 与作为子节点的结果。
## ResultPath 用于用结果更新输入中的节点
如果您为指定现有节点 ResultPath，则任务结果将替换该现有节点：
```
# State Input
{
"comment": "This is a test",
"details": "Default example",
"who" : "Step Functions"
}
# Path
"ResultPath": "$.comment"
# Task result
"Hello, Step Functions!"
# State Output
{
"comment": "Hello, Step Functions!",
"details": "Default example",
"who" : "Step Functions"
}
```
## ResultPath 用于在 a 中同时包含错误和输入 `Catch`
在某些情况下，您可能希望保留原始输入与错误。在 `Catch` 中使用 `ResultPath` 可包含错误与原始输入，而不是替换它。
```
"Catch": [{
"ErrorEquals": ["States.ALL"],
"Next": "NextTask",
"ResultPath": "$.error"
}]
```
如果上一 `Catch` 语句捕获错误，则它将在状态输入的 `error` 节点中包含结果。例如，对于以下输入：
```
{"foo": "bar"}
```
捕获错误时，状态输出为以下内容。
```
{
"foo": "bar",
"error": {
"Error": "Error here"
}
}
```
有关错误处理的更多信息，请参阅以下内容：
+ [处理 Step Functions 工作流程中的错误](concepts-error-handling.md)
+ [在 Step Functions 状态机中处理错误条件](tutorial-handling-error-conditions.md)
# 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
]
}
]
```