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

# ItemReader （地图）
<a name="input-output-itemreader"></a>

`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 字段的内容
<a name="itemreader-field-contents"></a>

根据您的数据集，`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 存储桶必须 Amazon Web Services 账户 与 Amazon Web Services 区域 您的状态机相同。  
*请注意，尽管您的状态机可能能够访问不同存储桶中相同 Amazon Web Services 区域存储桶中的文件 Amazon Web Services 账户 ，但 Step Functions 仅支持在 Amazon S3 存储桶中列出与状态机相同 Amazon Web Services 账户 和 Amazon Web Services 区域 相同的对象。*

## 处理嵌套数据集（2025 年 9 月 11 日更新）
<a name="itemreader-flatten"></a>

使用新 `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 %}"
            }
        },
        ...
    }
}
```

## 数据集示例
<a name="itemreader-examples-map"></a>

您可以指定下列选项之一作为数据集：
+ [上一个步骤中的 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 数据
<a name="itemsource-json-array"></a>

*分布式 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 对象列表
<a name="itemsource-example-s3-object-data"></a>

*分布式 Map 状态* 可以迭代存储在 Amazon S3 存储桶中的对象。**当工作流程执行达到`Map`状态时，Step Functions 会调用 [ListObjectsV2](https://docs.amazonaws.cn/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 存储桶中创建的每个**文件夹**都提供一个项目。文件夹项目会导致启动额外的子工作流执行。  
为避免为每个文件夹创建额外的子工作流程执行，我们建议您使用 Amazon CLI 来创建文件夹。有关更多信息，请参阅**《Amazon Command Line Interface 用户指南》中的[高级别 Amazon S3 命令](https://docs.amazonaws.cn/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 对象
<a name="itemsource-example-s3-object-data-flatten"></a>

通过增强对 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 文件
<a name="itemsource-example-json-data"></a>

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

当工作流程执行达到`Map`状态时，Step Functions 会调用 [GetObject](https://docs.amazonaws.cn/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 文件
<a name="itemsource-example-json-lines-data"></a>

*分布式 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 文件
<a name="itemsource-example-csv-data"></a>

**注意**  
使用 `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.amazonaws.cn/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 文件
<a name="itemsource-example-parquet-data"></a>

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 清单（处理多个项目）
<a name="itemsource-example-athena-manifest-data"></a>

您可以使用根据 `UNLOAD` 查询结果生成的 Athena 清单文件来指定 Map 状态的数据文件**来源**。可以将 `ManifestType` 设置为 `ATHENA_DATA`，将 `InputType` 设置为 `CSV`、`JSONL` 或 `Parquet`。

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

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

过程的概念性概述，简要说明如下：

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": "<S3_BUCKET_NAME>",
      "Key": "<S3_KEY_PREFIX><QUERY_ID>-manifest.csv"
   }
}
```

**在 Workflow Studio 中使用 Athena 清单模式**  
数据处理的常见场景是将 Map 应用于源自 Athena UNLOAD 查询的数据。Map 会调用 Lambda 函数来处理 Athena 清单中描述的每个项目。Step Functions Workflow Studio 提供了一种现成的模式，可将所有这些组件整合成一个块，您可以将其拖放到状态机画布上。

### S3 清单（处理多个项目）
<a name="itemsource-example-s3-inventory"></a>

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

当工作流程执行达到`Map`状态时，Step Functions 会调用 [GetObject](https://docs.amazonaws.cn/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.amazonaws.cn/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 策略建议
<a name="itemreader-iam-policies"></a>

当您使用 Step Functions 控制台创建工作流时，Step Functions 可以根据工作流定义中的资源自动生成 IAM 策略。生成的策略包括允许状态机角色调用*分布式地图状态*的 `[StartExecution](https://docs.amazonaws.cn/step-functions/latest/apireference/API_StartExecution.html)` API 操作和访问 Amazon 资源（例如 Amazon S3 存储桶和对象以及 Lambda 函数）所需的最低权限。

我们建议在 IAM 策略中仅包括必需的权限。例如，如果您的工作流包含分布式模式下的 `Map` 状态，则将策略范围缩小到包含您的数据的特定 Amazon S3 存储桶和文件夹。

**重要**  
如果您在*分布式 Map 状态* 输入中指定了 Amazon S3 存储桶和对象或前缀，并将[参考路径](amazon-states-language-paths.md#amazon-states-language-reference-paths)指向现有键值对，请务必更新工作流程的 IAM 策略。将策略范围缩小到运行时该路径解析到的存储桶和对象名称。

以下示例展示了使用 [ListObjectsV2](https://docs.amazonaws.cn/AmazonS3/latest/API/API_ListObjectsV2.html) 和 [GetObject](https://docs.amazonaws.cn/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.amazonaws.cn/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` 不会限定范围，您将为对象使用通配符 (`*`)。