ItemReader (Peta)

Status Peta Terdistribusi dapat menerima input JSON yang diteruskan dari langkah sebelumnya dalam alur kerja.

Input dapat berupa array JSON, objek JSON, atau array dalam node dari objek JSON.

Step Functions akan iterasi langsung di atas elemen array, atau pasangan kunci-nilai dari objek JSON.

Untuk memilih node tertentu yang berisi array dari objek JSON, Anda dapat menggunakan ItemsPath (Peta, JSONPath hanya) atau menggunakan JSONata ekspresi di Items bidang untuk JSONata status.

Untuk memproses item individual, status Peta Terdistribusi memulai eksekusi alur kerja anak untuk setiap item. Tab berikut menunjukkan contoh input yang diteruskan ke Map status dan input yang sesuai ke eksekusi alur kerja anak.

Status Peta Terdistribusi dapat mengulang objek yang disimpan dalam bucket Amazon S3. Saat eksekusi alur kerja mencapai Map status, Step Functions memanggil aksi ListObjectsV2 API, yang mengembalikan array metadata objek Amazon S3. Dalam array ini, setiap item berisi data, seperti ETagdan Key, untuk data aktual yang disimpan dalam bucket.

Untuk memproses item individual dalam array, status Peta Terdistribusi memulai eksekusi alur kerja anak. Misalnya, ember Amazon S3 Anda berisi 100 gambar. Kemudian, array yang dikembalikan setelah menjalankan aksi ListObjectsV2 API berisi 100 item metadata. Status Peta Terdistribusi kemudian memulai 100 eksekusi alur kerja anak untuk memproses setiap item.

Untuk memproses objek data secara langsung, tanpa alur kerja bersarang, Anda dapat memilih opsi Transformasi LOAD_AND_FLATTEN untuk memproses item secara langsung.

Tab berikut menunjukkan contoh sintaks ItemReader bidang dan masukan yang diteruskan ke eksekusi alur kerja anak untuk kumpulan data ini.

Dengan dukungan yang disempurnakan untuk S3 ListObjects V2 sebagai sumber input di Peta Terdistribusi, mesin status Anda dapat membaca dan memproses beberapa objek data dari bucket Amazon S3 secara langsung, sehingga tidak perlu peta bersarang untuk memproses metadata!

Dengan LOAD_AND_FLATTEN opsi tersebut, mesin negara Anda akan melakukan hal berikut:

Dengan opsi transformasi, Anda tidak lagi memerlukan Peta Terdistribusi bersarang untuk memproses metadata. Menggunakan opsi LOAD_AND_FLATTEN meningkatkan skalabilitas, mengurangi jumlah run map aktif, dan memproses beberapa objek secara bersamaan.

Konfigurasi berikut menunjukkan pengaturan untukItemReader:

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

Status Peta Terdistribusi dapat menerima file JSON yang disimpan di bucket Amazon S3 sebagai kumpulan data. File JSON harus berisi array atau objek JSON.

Ketika eksekusi alur kerja mencapai Map status, Step Functions memanggil aksi GetObjectAPI untuk mengambil file JSON yang ditentukan.

Jika file JSON berisi struktur objek bersarang, Anda dapat memilih node tertentu dengan kumpulan data Anda dengan file. ItemsPointer Misalnya, konfigurasi berikut akan mengekstrak daftar bersarang produk unggulan dalam inventaris.

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

MapStatus kemudian mengulangi setiap item dalam array dan memulai eksekusi alur kerja anak untuk setiap item. Misalnya, jika file JSON Anda berisi 1000 item array, Map status memulai 1000 eksekusi alur kerja anak.

Tab berikut menunjukkan contoh sintaks ItemReader bidang dan masukan yang diteruskan ke eksekusi alur kerja anak untuk kumpulan data ini.

Untuk contoh ini, bayangkan Anda memiliki file JSON bernamafactcheck.json. Anda telah menyimpan file ini dalam awalan bernama jsonDataset dalam bucket Amazon S3. Berikut ini adalah contoh dari dataset 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"
  },
  ...
]

Status Peta Terdistribusi dapat menerima file JSON Lines yang disimpan di bucket Amazon S3 sebagai kumpulan data.

Tab berikut menunjukkan contoh sintaks ItemReader bidang dan masukan yang diteruskan ke eksekusi alur kerja anak untuk kumpulan data ini.

Untuk contoh ini, bayangkan Anda memiliki file JSON Lines bernamafactcheck.jsonl. Anda telah menyimpan file ini dalam awalan bernama jsonlDataset dalam bucket Amazon S3. Berikut ini adalah contoh isi file.

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

Status Peta Terdistribusi dapat menerima file teks yang dibatasi yang disimpan dalam bucket Amazon S3 sebagai kumpulan data. Jika Anda menggunakan file yang dibatasi teks sebagai kumpulan data Anda, Anda perlu menentukan header kolom. Untuk informasi tentang cara menentukan header, lihatIsi ItemReader lapangan.

Step Functions mem-parsing file teks yang dibatasi berdasarkan aturan berikut:

Untuk informasi selengkapnya tentang cara Step Functions mem-parsing file teks yang dibatasi, lihat. Example of parsing an input CSV file

Saat eksekusi alur kerja mencapai Map status, Step Functions memanggil aksi GetObjectAPI untuk mengambil file yang ditentukan. MapStatus kemudian mengulangi setiap baris dalam file dan memulai eksekusi alur kerja anak untuk memproses item di setiap baris. Misalnya, misalkan Anda menyediakan file teks yang dibatasi yang berisi 100 baris sebagai input. Kemudian, penerjemah meneruskan setiap baris ke Map negara bagian. MapStatus memproses item dalam urutan serial, dimulai setelah baris header.

Tab berikut menunjukkan contoh sintaks ItemReader bidang dan masukan yang diteruskan ke eksekusi alur kerja anak untuk kumpulan data ini.

File parket dapat digunakan sebagai sumber input. File Apache Parquet yang disimpan di Amazon S3 menyediakan pemrosesan data kolumnar yang efisien dalam skala besar.

Saat menggunakan file Parket, ketentuan berikut berlaku:

Berikut ini menunjukkan contoh konfigurasi ASL untuk InputType diatur ke Parket:

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

Anda dapat menggunakan file manifes Athena, yang dihasilkan dari hasil UNLOAD kueri, untuk menentukan sumber file data untuk status Peta Anda. Anda mengatur ManifestType keATHENA_DATA, dan InputType ke salah satuCSV,JSONL, atauParquet.

Saat menjalankan UNLOAD kueri, Athena menghasilkan file manifes data selain objek data aktual. File manifes menyediakan daftar CSV terstruktur dari file data. File manifes dan data disimpan ke lokasi hasil kueri Athena Anda di Amazon S3.

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

Tinjauan konseptual proses, secara singkat:

Fitur ini dapat memproses format keluaran CSV, JSONL, dan Parket dari Athena. Semua objek yang direferensikan dalam satu file manifes harus InputType format yang sama. Perhatikan bahwa objek CSV yang diekspor oleh UNLOAD kueri tidak menyertakan header di baris pertama. Lihat CSVHeaderLocation apakah Anda perlu memberikan header kolom.

Konteks peta juga akan menyertakan $states.context.Map.Item.Source sehingga Anda dapat menyesuaikan pemrosesan berdasarkan sumber data.

Berikut ini adalah contoh konfigurasi yang ItemReader dikonfigurasi untuk menggunakan manifes Athena:

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

Status Peta Terdistribusi dapat menerima file manifes inventaris Amazon S3 yang disimpan di bucket Amazon S3 sebagai kumpulan data.

Saat eksekusi alur kerja mencapai Map status, Step Functions akan memanggil tindakan GetObjectAPI untuk mengambil file manifes inventaris Amazon S3 yang ditentukan.

Secara default, Map status kemudian mengulangi objek dalam inventaris untuk mengembalikan array metadata objek inventaris Amazon S3.

Jika Anda menentukan ManifestType adalah S3_INVENTORY maka InputType tidak dapat ditentukan.

Berikut ini adalah contoh file inventaris dalam format CSV. File ini menyertakan objek bernama csvDataset danimageDataset, yang disimpan dalam bucket Amazon S3 yang diberi nama. amzn-s3-demo-source-bucket

"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"
...

Contoh berikut dari file manifes inventaris Amazon S3 menunjukkan header CSV untuk metadata objek inventaris.

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

Tab berikut menunjukkan contoh sintaks ItemReader bidang dan masukan yang diteruskan ke eksekusi alur kerja anak untuk kumpulan data ini.