ItemReader (Map)

分散マップ状態は、ワークフローの前のステップから渡された JSON 入力を受け入れることができます。

入力は JSON 配列、JSON オブジェクト、または JSON オブジェクトのノード内の配列のいずれかにできます。

Step Functions は、配列の要素、または JSON オブジェクトのキーと値のペアを直接反復処理します。

JSON オブジェクトから配列を含む特定のノードを選択するには、ItemsPath (マップ、JSONPath のみ) を使用するか、JSONata ステートの場合は Items フィールドで JSONata 式を使用できます。

個々の項目を処理するために、Distributed Map ステートは項目ごとに子ワークフロー実行を開始します。以下のタブには、Map 状態に渡される入力と、それに対応する子ワークフロー実行への入力の例が表示されます。

分散マップ状態では、Amazon S3 バケットに格納されているオブジェクトを反復できます。ワークフロー実行が Map ステートに到達すると、Step Functions は ListObjectsV2 API アクションを呼び出し、Amazon S3 オブジェクトメタデータの配列を返します。この配列の各項目には、バケットに保存されている実際のデータの [ETag] や [Key] などのデータが含まれています。

配列内の個々の項目を処理するために、分散マップ状態は子ワークフロー実行を開始します。例えば、Amazon S3 バケットに 100 個のイメージが含まれているとします。次に、ListObjectsV2 API アクションを呼び出した後に返される配列には 100 個のメタデータ項目が含まれています。Distributed Map ステートは 100 個の子ワークフロー実行を開始して各項目を処理します。

ネストされたワークフローを使用せずにデータオブジェクトを直接処理するには、LOAD_AND_FLATTEN 変換オプションを選択して項目を直接処理できます。

以下のタブは、このデータセットの子ワークフロー実行に渡される ItemReader フィールド構文と入力の例を示しています。

分散マップの入力ソースとしての S3 ListObjectsV2 のサポートの強化により、ステートマシンは Amazon S3 バケットから複数のデータオブジェクトを直接読み取って処理できるようになり、メタデータ処理のためのネストされたマップが不要になります。

LOAD_AND_FLATTEN オプションを使用すると、ステートマシンは次の処理を行います。

この変換オプションにより、メタデータの処理にネストされた分散マップは不要になります。LOAD_AND_FLATTEN オプションを使用すると、スケーラビリティの向上、アクティブ状態のマップ実行数の削減、複数のオブジェクトの同時処理が行われます。

次に示しているのは、ItemReader の設定です。

"ItemReader": {
   "Resource": "arn:aws:states:::s3:listObjectsV2",
   "ReaderConfig": {
      "InputType": "JSON",
      "Transformation": "LOAD_AND_FLATTEN"
   },
   "Arguments": {
      "Bucket": "S3_BUCKET_NAME",
      "Prefix": "S3_BUCKET_PREFIX"
   }
}

分散マップ状態は、Amazon S3 バケットにデータセットとして保存されている JSON ファイルを受け入れることができます。JSON ファイルには配列または JSON オブジェクトが含まれている必要があります。

ワークフローの実行が Map 状態に達すると、Step Functions は GetObject API アクションを呼び出して、指定された JSON ファイルを取得します。

JSON ファイルにネストされたオブジェクト構造が含まれている場合は、ItemsPointer を使用して、データセットを含む特定のノードを選択できます。例えば、次の設定では、inventory 内の featured products のネストされたリストが抽出されます。

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "JSON",
      "ItemsPointer": "/inventory/products/featured"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "nested-data-file.json"
   }
}

その後、Map 状態は配列内の各項目を反復処理し、項目ごとに子ワークフローの実行を開始します。例えば、JSON ファイルに 1000 個の配列項目が含まれている場合、Map 状態は 1000 個の子ワークフロー実行を開始します。

以下のタブは、このデータセットの子ワークフロー実行に渡される ItemReader フィールド構文と入力の例を示しています。

この例では、factcheck.json という JSON ファイルがあるとします。このファイルを Amazon S3 バケットに jsonDataset という名前のプレフィックスで保存しました。以下は、JSON データセットの例です。

[
  {
    "verdict": "true",
    "statement_date": "6/11/2008",
    "statement_source": "speech"
  },
  {
    "verdict": "false",
    "statement_date": "6/7/2022",
    "statement_source": "television"
  },
  {
    "verdict": "mostly-true",
    "statement_date": "5/18/2016",
    "statement_source": "news"
  },
  ...
]

Distributed Map ステートは、Amazon S3 バケットに保存されている JSON Lines ファイルをデータセットとして受け入れることができます。

以下のタブは、このデータセットの子ワークフロー実行に渡される ItemReader フィールド構文と入力の例を示しています。

この例では、factcheck.jsonl という JSON Lines ファイルがあるとします。このファイルを Amazon S3 バケットに jsonlDataset という名前のプレフィックスで保存しました。次に示しているのは、そのファイルの内容の例です。

{"verdict": "true", "statement_date": "6/11/2008", "statement_source": "speech"} 
{"verdict": "false", "statement_date": "6/7/2022", "statement_source": "television"}
{"verdict": "mostly-true", "statement_date": "5/18/2016", "statement_source": "news"}

Distributed Map ステートは、Amazon S3 バケットにデータセットとして保存されているテキスト区切りファイルを受け入れることができます。テキスト区切りファイルをデータセットとして使用する場合は、列ヘッダーを指定する必要があります。ヘッダーの指定方法については、「ItemReader フィールドの内容」を参照してください。

Step Functions は、次のルールに基づいてテキスト区切りファイルを解析します。

Step Functions がテキスト区切りファイルを解析する方法の詳細については、「Example of parsing an input CSV file」を参照してください。

ワークフロー実行が Map ステートに到達すると、Step Functions は GetObject API アクションを呼び出して、指定されたファイルを取得します。その後、Map ステートはファイル内の各行を反復処理し、各行の項目を処理するために子ワークフロー実行を開始します。例えば、100 行を含むテキスト区切りファイルを入力として用意したとします。そうすると、インタープリタは各行を Map 状態に渡します。Map 状態は項目をヘッダー行の後から順次処理します。

以下のタブは、このデータセットの子ワークフロー実行に渡される ItemReader フィールド構文と入力の例を示しています。

Parquet ファイルは入力ソースとして使用できます。Amazon S3 に保存した Apache Parquet ファイルにより、大規模で効率的な列指向データ処理が可能になります。

Parquet ファイルを使用する場合、次の条件が適用されます。

次に示しているのは、InputType に Parquet を指定した場合の ASL の設定例です。

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "PARQUET"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "my-parquet-data-file-1.parquet"
   }
}

UNLOAD クエリ結果から生成された Athena マニフェストファイルを使用して、Map ステートのデータファイルのソースを指定できます。ManifestType を ATHENA_DATA に設定し、InputType を CSV、JSONL、または Parquet に設定します。

UNLOAD クエリを実行すると、Athena は実際のデータオブジェクトに加えて、データマニフェストファイルを生成します。マニフェストファイルには、データファイルの構造化された CSV リストが含まれます。マニフェストとデータファイルはどちらも、Athena のクエリ結果の保存先である Amazon S3 に保存されます。

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

プロセスの概念的概要 (要約):

この機能は、Athena からの CSV、JSONL、および Parquet 出力形式を処理できます。1 つのマニフェストファイルで参照されるすべてのオブジェクトは、同じ InputType 形式である必要があります。UNLOAD クエリによってエクスポートされた CSV オブジェクトでは、最初の行にヘッダーが含まれていないことに注意してください。列ヘッダーを指定する必要がある場合は、「CSVHeaderLocation」を参照してください。

マップのコンテキストには $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"
   }
}

分散マップ状態は、Amazon S3 バケットにデータセットとして保存されている Amazon S3 インベントリマニフェストファイルを受け入れることができます。

ワークフローの実行が Map 状態に達すると、Step Functions は GetObject API アクションを呼び出して、指定された Amazon S3 インベントリマニフェストファイルを取得します。

デフォルトでは、Map ステートはインベントリ内のオブジェクトを反復処理し、Amazon S3 インベントリオブジェクトのメタデータ配列を返します。

ManifestType を S3_INVENTORY に指定した場合、InputType を指定することはできません。

CSV 形式のインベントリファイルの例を次に示します。このファイルには imageDataset および csvDataset という名前のオブジェクトが含まれます。これらのオブジェクトは、amzn-s3-demo-source-bucket という名前の Amazon S3 バケットに保存されます。

"amzn-s3-demo-source-bucket","csvDataset/","0","2022-11-16T00:27:19.000Z"
"amzn-s3-demo-source-bucket","csvDataset/titles.csv","3399671","2022-11-16T00:29:32.000Z"
"amzn-s3-demo-source-bucket","imageDataset/","0","2022-11-15T20:00:44.000Z"
"amzn-s3-demo-source-bucket","imageDataset/n02085620_10074.jpg","27034","2022-11-15T20:02:16.000Z"
...

次の Amazon S3 インベントリマニフェストファイルの例は、インベントリオブジェクトメタデータの CSV ヘッダーを示しています。

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

以下のタブは、このデータセットの子ワークフロー実行に渡される ItemReader フィールド構文と入力の例を示しています。