ItemReader

ItemReader フィールドは JSON オブジェクトで、データセットとその場所を指定します。分散マップ状態はこのデータセットを入力として使用します。次の例は、データセットが Amazon S3 バケットに保存されている CSV ファイルである場合の ItemReader フィールドの構文を示しています。

"ItemReader": {
    "ReaderConfig": {
        "InputType": "CSV",
        "CSVHeaderLocation": "FIRST_ROW"
    },
    "Resource": "arn:aws:states:::s3:getObject",
    "Parameters": {
        "Bucket": "myBucket",
        "Key": "csvDataset/ratings.csv"
    }
}

ヒント

Workflow Studio では、データセットとその場所を [アイテムソース] フィールドで指定します。

ItemReader フィールドの内容

データセットによって、ItemReader フィールドの内容は異なります。例えば、データセットがワークフローの前のステップから渡された JSON 配列である場合、ItemReader フィールドは省略されます。データセットが Amazon S3 データソースである場合、このフィールドには次のサブフィールドが含まれます。

ReaderConfig

以下の詳細を指定する JSON オブジェクト。

• InputType

  CSV ファイル、オブジェクト、JSON ファイル、Amazon S3 インベントリリストなど、Amazon S3 データソースのタイプを指定します。Workflow Studio では、[項目ソース] フィールドの下にある [Amazon S3 項目ソース] ドロップダウンリストから入力タイプを選択できます。

• CSVHeaderLocation

  注記

  このフィールドは、CSV ファイルをデータセットとして使用する場合にのみ指定する必要があります。

  以下の値のいずれかを受け入れ、列ヘッダーの位置を指定します。

  重要

  現在、Step Functions は最大 10 KB の CSV ヘッダーをサポートしています。

  • FIRST_ROW - ファイルの最初の行がヘッダーの場合は、このオプションを使用します。

  • GIVEN - このオプションを使用して、ステートマシン定義内のヘッダーを指定します。例えば、CSV ファイルに次のデータが含まれている場合を考えてみましょう。

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

  ヒント

  Workflow Studio では、このオプションは [項目ソース] フィールドの [追加設定] にあります。

• MaxItems

  Map 状態に渡されるデータ項目の数を制限します。例えば、1000 行の CSV ファイルを用意し、100 という制限を指定したとします。そうすると、インタープリタは 100 行のみを Map 状態に渡します。Map 状態は項目をヘッダー行の後から順番に処理します。

  デフォルトでは、Map 状態は指定されたデータセット内のすべての項目を反復処理します。

  注記

  現在、100,000,000 件の制限まで指定できます。分散マップ状態は、この制限を超える項目の読み取りを停止します。

  ヒント

  Workflow Studio では、このオプションは [項目ソース] フィールドの [追加設定] にあります。

  または、分散マップ状態の入力に既存のキーと値のペアへの参照パスを指定することもできます。このパスは正の整数に変換する必要があります。MaxItemsPath サブフィールドで参照パスを指定します。

  重要

  MaxItems または MaxItemsPath サブフィールドのいずれかを指定できますが、両方を指定することはできません。

Resource

指定されたデータセットに応じて、Step Functions が呼び出す必要がある Amazon S3 API アクション。

Parameters

データセットが保存されている Amazon S3 バケットの名前とオブジェクトキーを指定する JSON オブジェクト。

重要

Amazon S3 バケットがステートマシンと同じ AWS アカウント と AWS リージョン にあることを確認してください。

データセットの例

データセットとして以下のいずれかのオプションを指定できます。

• 前のステップの JSON 配列

• Amazon S3 オブジェクトのリスト

• Amazon S3 バケット内の JSON ファイル

• Amazon S3 バケット内の CSV ファイル

• Amazon S3 インベントリリスト

重要

Step Functions には、使用する Amazon S3 データセットにアクセスするための適切な許可が必要です。データセットの IAM ポリシーの詳細については、「データセットの IAM ポリシー」を参照してください。

前のステップの JSON 配列

分散マップ状態は、ワークフローの前のステップから渡された JSON 入力を受け入れることができます。この入力は配列であるか、特定のノード内に配列を含んでいる必要があります。配列を含むノードを選択するには、ItemsPath フィールドを使用できます。

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

注記

Step Functions は、データセットが前のステップの JSON 配列の場合、ItemReader フィールドを省略します。

Input passed to the Map state

次の 3 つの項目からなる 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

分散マップ状態は、3 つの子ワークフローの実行を開始します。実行するたびに、配列項目を入力として受け取ります。次の例は、子ワークフロー実行が受け取る入力を示しています。

{
  "verdict": "true",
  "statement_date": "6/11/2008",
  "statement_source": "speech"
}

Amazon S3 オブジェクトの例

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

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

注記

• 現在、Step Functions には、Amazon S3 コンソールを使用して特定の Amazon S3 バケットに作成する各フォルダの項目も含まれています。その結果、分散マップ状態によって開始される、余分の子ワークフロー実行が生じます。フォルダに余分な子ワークフロー実行が作成されないようにするため、AWS CLI を使用してフォルダを作成することをお勧めします。詳細については、「AWS Command Line Interface ユーザーガイド」の「高レベルな S3 コマンド」を参照してください。

• Step Functions には、使用する Amazon S3 データセットにアクセスするための適切な許可が必要です。データセットの IAM ポリシーの詳細については、「データセットの IAM ポリシー」を参照してください。

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

ItemReader syntax

この例では、イメージ、JSON ファイル、オブジェクトを含むデータを、myBucket という名前の Amazon S3 バケットの processData というプレフィックス内に整理しました。

"ItemReader": {
    "Resource": "arn:aws:states:::s3:listObjectsV2",
    "Parameters": {
        "Bucket": "myBucket",
        "Prefix": "processData"
    }
}

Input passed to a child workflow execution

分散マップ状態は、Amazon S3 バケットに存在する項目数と同じ数の子ワークフローの実行を開始します。次の例は、子ワークフロー実行が受け取る入力を示しています。

{
  "Etag": "\"05704fbdccb224cb01c59005bebbad28\"",
  "Key": "processData/images/n02085620_1073.jpg",
  "LastModified": 1668699881,
  "Size": 34910,
  "StorageClass": "STANDARD"
}

Amazon S3 バケット内の JSON ファイル

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

ワークフローの実行が Map 状態に達すると、Step Functions は GetObject API アクションを呼び出して、指定された JSON ファイルを取得します。その後、Map 状態は配列内の各項目を反復処理し、項目ごとに子ワークフローの実行を開始します。例えば、JSON ファイルに 1000 個の配列項目が含まれている場合、Map 状態は 1000 個の子ワークフロー実行を開始します。

注記

• 子ワークフロー実行を開始するために使用される実行入力は、256 KB を超えることができません。ただし、Step Functions では、オプションの ItemSelector フィールドを適用して項目のサイズを小さくすると、CSV または JSON ファイルから最大 8 MB の項目を読み取ることができます。

• 現在、Step Functions は Amazon S3 インベントリレポートの個々のファイルに対して 10 GB の最大サイズをサポートしています。ただし、Step Functions は、個々のファイルが 10 GB 未満の場合、10 GB 以上を処理できます。

• Step Functions には、使用する Amazon S3 データセットにアクセスするための適切な許可が必要です。データセットの IAM ポリシーの詳細については、「データセットの IAM ポリシー」を参照してください。

以下のタブは、このデータセットの子ワークフロー実行に渡される 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": "myBucket",
        "Key": "jsonDataset/factcheck.json"
    }
}

Input to a child workflow execution

分散マップ状態は、Json ファイルに存在する配列項目数と同じ数の子ワークフローの実行を開始します。次の例は、子ワークフロー実行が受け取る入力を示しています。

{
  "verdict": "true",
  "statement_date": "6/11/2008",
  "statement_source": "speech"
}

Amazon S3 バケット内の CSV ファイル

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

CSV ファイルのデータを作成および管理するための標準化された形式がないため、Step Functions は次のルールに基づいて CSV ファイルを解析します。

• カンマ (,) は個々のフィールドを区切る区切り文字です。

• 改行は個々のレコードを分割する区切り文字です。

• フィールドは文字列として扱われます。データ型変換には、ItemSelector の States.StringToJson 組み込み関数を使用します。

• 文字列を二重引用符 (" ") で囲む必要はありません。ただし、二重引用符で囲まれた文字列には、区切り文字として機能しないカンマや改行を含めることができます。

• 二重引用符は、繰り返しによりエスケープできます。

• 行のフィールド数がヘッダーのフィールド数より少ない場合、Step Functions は欠落している値に空の文字列を提供します。

• 行のフィールド数がヘッダーのフィールド数よりも多い場合、Step Functions は追加のフィールドをスキップします。

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

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

注記

• 子ワークフロー実行を開始するために使用される実行入力は、256 KB を超えることができません。ただし、Step Functions では、オプションの ItemSelector フィールドを適用して項目のサイズを小さくすると、CSV または JSON ファイルから最大 8 MB の項目を読み取ることができます。

• 現在、Step Functions は Amazon S3 インベントリレポートの個々のファイルに対して 10 GB の最大サイズをサポートしています。ただし、Step Functions は、個々のファイルが 10 GB 未満の場合、10 GB 以上を処理できます。

• Step Functions には、使用する Amazon S3 データセットにアクセスするための適切な許可が必要です。データセットの IAM ポリシーの詳細については、「データセットの IAM ポリシー」を参照してください。

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

ItemReader syntax

例えば、ratings.csv という名前の CSV ファイルがあるとします。次に、このファイルを Amazon S3 バケットに csvDataset という名前のプレフィックスで保存しました。

{
  "ItemReader": {
    "ReaderConfig": {
      "InputType": "CSV",
      "CSVHeaderLocation": "FIRST_ROW"
    },
    "Resource": "arn:aws:states:::s3:getObject",
    "Parameters": {
      "Bucket": "myBucket",
      "Key": "csvDataset/ratings.csv"
    }
  }
}

Input to a child workflow execution

分散マップ状態は、ヘッダー行を除いて (ファイル内にある場合)、CSV ファイルに存在する行数と同じ数の子ワークフローの実行を開始します。次の例は、子ワークフロー実行が受け取る入力を示しています。

{
  "rating": "3.5",
  "movieId": "307",
  "userId": "1",
  "timestamp": "1256677221"
}

S3 インベントリの例

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

ワークフローの実行が Map 状態に達すると、Step Functions は GetObject API アクションを呼び出して、指定された Amazon S3 インベントリマニフェストファイルを取得します。その後、Map 状態はインベントリ内のオブジェクトを反復して、Amazon S3 インベントリオブジェクトメタデータの配列を返します。

注記

• 現在、Step Functions は Amazon S3 インベントリレポートの個々のファイルに対して 10 GB の最大サイズをサポートしています。ただし、Step Functions は、個々のファイルが 10 GB 未満の場合、10 GB 以上を処理できます。

• Step Functions には、使用する Amazon S3 データセットにアクセスするための適切な許可が必要です。データセットの IAM ポリシーの詳細については、「データセットの IAM ポリシー」を参照してください。

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

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

重要

現在、Step Functions はユーザー定義の Amazon S3 インベントリレポートをデータセットとしてサポートしていません。また、Amazon S3 インベントリレポートの出力形式が CSV であることを確認する必要があります。Amazon S3 インベントリとその設定方法の詳細については、「Amazon S3 ユーザーガイド」の「Amazon S3 インベントリ」を参照してください。

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

{
  "sourceBucket" : "sourceBucket",
  "destinationBucket" : "arn:aws:s3:::inventory",
  "version" : "2016-11-30",
  "creationTimestamp" : "1668560400000",
  "fileFormat" : "CSV",
  "fileSchema" : "Bucket, Key, Size, LastModifiedDate",
  "files" : [ {
    "key" : "source-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": "destinationBucket",
      "Key": "destination-prefix/source-bucket/config-ID/YYYY-MM-DDTHH-MMZ/manifest.json"
    }
  }
}

Input to a child workflow execution

{
  "LastModifiedDate": "2022-11-16T00:29:32.000Z",
  "Bucket": "sourceBucket",
  "Size": "3399671",
  "Key": "csvDataset/titles.csv"
}

Amazon S3 インベントリレポートの設定時に選択したフィールドによっては、示されている例と manifest.json ファイルの内容が異なる場合があります。

データセットの IAM ポリシー

Step Functions コンソールでワークフローを作成すると、Step Functions はワークフロー定義内のリソースに基づいて IAM ポリシーを自動的に生成できます。これらのポリシーには、ステートマシンロールが分散マップ状態の StartExecution API アクションを呼び出すために必要な最小特権が含まれています。これらのポリシーには、Amazon S3 バケットやオブジェクト、Lambda 関数などの AWS リソースへのアクセスに必要な最小特権を持つ Step Functions も含まれています。IAM ポリシーには必要なアクセス許可のみを含めることを強くお勧めします。例えばワークフローに分散モードの Map 状態が含まれている場合は、ポリシーの範囲をデータセットを含む特定の Amazon S3 バケットとフォルダに限定します。

重要

分散マップ状態の入力で、既存のキーと値のペアへの参照パスとともに、Amazon S3 バケットやオブジェクト、またはプレフィックスを指定する場合は、ワークフローの IAM ポリシーを必ず更新してください。ポリシーの範囲は、ランタイムでパスから解釈されるバケット名とオブジェクト名に限定します。

次の IAM ポリシーの例では、ListObjectSv2 および GetObject API アクションを使用して Amazon S3 データセットにアクセスするのに必要な最小特権を付与しています。

例 データセットとしての Amazon S3 オブジェクトの IAM ポリシー

以下は、Amazon S3 バケットの processImages に配置した myBucket という名前のオブジェクトにアクセスするための最小特権を付与する IAM ポリシーの例です。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::myBucket"
            ],
            "Condition": {
                "StringLike": {
                    "s3:prefix": [
                        "processImages"
                    ]
                }
            }
        }
    ]
}

例 データセットとしての CSV ファイルの IAM ポリシー

以下は、ratings.csv という名前の CSV ファイルにアクセスするための最小特権を付与する IAM ポリシーの例です。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::myBucket/csvDataset/ratings.csv"
            ]
        }
    ]
}

例 データセットとしての Amazon S3 インベントリの IAM ポリシー

以下は、Amazon S3 インベントリレポートにアクセスするための最小特権を付与する IAM ポリシーの例です。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::destination-prefix/source-bucket/config-ID/YYYY-MM-DDTHH-MMZ/manifest.json",
                "arn:aws:s3:::destination-prefix/source-bucket/config-ID/data/*"
            ]
        }
    ]
}