ItemReader (Map)

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

入力は、JSON 配列、JSON オブジェクト、または JSON オブジェクトのノード内の配列です。

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

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

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

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

配列内の個々の項目を処理するために、分散マップ状態は子ワークフロー実行を開始します。例えば、Amazon S3 バケットに 100 個のイメージが含まれているとします。次に、ListObjectsV2API アクションを呼び出した後に返される配列には、100 個のメタデータ項目が含まれます。その後、分散マップ状態は 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。たとえば、次の設定では、インベントリ内の注目製品のネストされたリストが抽出されます。

"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"
  },
  ...
]

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

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

この例では、 という名前の JSON Lines ファイルがあるとしますfactcheck.jsonl。このファイルを 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"}

分散マップ状態は、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 ファイルを使用する場合、次の条件が適用されます。

以下は、 を Parquet InputTypeに設定する 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 マニフェストファイルを使用して、マップ状態のデータファイルのソースを指定できます。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')

プロセスの概念的な概要を簡単に説明します。

この機能では、Athena から CSV、JSONL、および Parquet の出力形式を処理できます。単一のマニフェストファイルで参照されるすべてのオブジェクトは、同じ 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 フィールド構文と入力の例を示しています。