ItemReader (Mappa)

Uno stato della mappa distribuita può accettare un input JSON passato da una fase precedente del flusso di lavoro.

L'input può essere un array JSON, un oggetto JSON o un array all'interno di un nodo di un oggetto JSON.

Step Functions itererà direttamente sugli elementi di un array o sulle coppie chiave-valore di un oggetto JSON.

Per selezionare un nodo specifico che contiene un array da un oggetto JSON, puoi usare ItemsPath ( JSONPath Solo mappa) o usare un' JSONata espressione nel campo per gli stati. Items JSONata

Per elaborare singoli elementi, lo stato Distributed Map avvia un'esecuzione del flusso di lavoro secondario per ogni elemento. Le schede seguenti mostrano esempi dell'input passato Map allo stato e dell'input corrispondente all'esecuzione di un flusso di lavoro secondario.

Uno stato della mappa distribuita può iterare sugli oggetti archiviati in un bucket Amazon S3. Quando l'esecuzione del flusso di lavoro raggiunge lo Map stato, Step Functions richiama l'azione API ListObjectsV2, che restituisce un array di metadati degli oggetti Amazon S3. In questo array, ogni elemento contiene dati, ad esempio una chiave, per i ETagdati effettivi memorizzati nel bucket.

Per elaborare singoli elementi dell'array, lo stato Distributed Map avvia l'esecuzione di un workflow secondario. Ad esempio, supponiamo che il bucket Amazon S3 contenga 100 immagini. Quindi, l'array restituito dopo aver richiamato l'azione ListObjectsV2 API contiene 100 elementi di metadati. Lo stato Distributed Map avvia quindi 100 esecuzioni di workflow secondarie per elaborare ogni elemento.

Per elaborare gli oggetti di dati direttamente, senza flussi di lavoro annidati, potete scegliere l'opzione di trasformazione LOAD_AND_FLATTEN per elaborare direttamente gli elementi.

Le schede seguenti mostrano esempi della sintassi del ItemReader campo e dell'input passato all'esecuzione di un flusso di lavoro secondario per questo set di dati.

Con il supporto avanzato per S3 ListObjects V2 come fonte di input in Distributed Map, le tue macchine a stati possono leggere ed elaborare più oggetti di dati direttamente dai bucket Amazon S3, eliminando la necessità di mappe annidate per elaborare i metadati!

Con l'LOAD_AND_FLATTENopzione, la tua macchina a stati eseguirà le seguenti operazioni:

Con l'opzione di trasformazione, non sono più necessarie mappe distribuite annidate per elaborare i metadati. L'utilizzo dell'opzione LOAD_AND_FLATTEN aumenta la scalabilità, riduce il numero di esecuzioni di mappe attive ed elabora più oggetti contemporaneamente.

La configurazione seguente mostra l'impostazione per un: ItemReader

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

Uno stato della mappa distribuita può accettare un file JSON archiviato in un bucket Amazon S3 come set di dati. Il file JSON deve contenere un array o un oggetto JSON.

Quando l'esecuzione del flusso di lavoro raggiunge lo Map stato, Step Functions richiama l'azione GetObjectAPI per recuperare il file JSON specificato.

Se il file JSON contiene una struttura a oggetti annidata, puoi selezionare il nodo specifico con il set di dati con un. ItemsPointer Ad esempio, la seguente configurazione consente di estrarre un elenco annidato di prodotti in evidenza nell'inventario.

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

Lo Map stato esegue quindi un'iterazione su ogni elemento dell'array e avvia un'esecuzione del flusso di lavoro secondario per ogni elemento. Ad esempio, se il file JSON contiene 1000 elementi dell'array, lo Map stato avvia 1000 esecuzioni di workflow secondarie.

Le schede seguenti mostrano esempi della sintassi del ItemReader campo e dell'input passato all'esecuzione di un flusso di lavoro secondario per questo set di dati.

Per questo esempio, immaginate di avere un file JSON denominato. factcheck.json Hai memorizzato questo file all'interno di un prefisso denominato jsonDataset in un bucket Amazon S3. Di seguito è riportato un esempio del set di dati 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"
  },
  ...
]

Uno stato della mappa distribuita può accettare un file JSON Lines archiviato in un bucket Amazon S3 come set di dati.

Le schede seguenti mostrano esempi della sintassi del ItemReader campo e dell'input passato all'esecuzione di un flusso di lavoro secondario per questo set di dati.

Per questo esempio, immaginate di avere un file JSON Lines denominato. factcheck.jsonl Hai memorizzato questo file all'interno di un prefisso denominato jsonlDataset in un bucket Amazon S3. Di seguito è riportato un esempio del contenuto del 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"}

Uno stato della mappa distribuita può accettare un file delimitato da testo archiviato in un bucket Amazon S3 come set di dati. Se utilizzi un file delimitato da testo come set di dati, devi specificare un'intestazione di colonna. Per informazioni su come specificare un'intestazione, consulta. Contenuto del ItemReader campo

Step Functions analizza i file delimitati da testo in base alle seguenti regole:

Per ulteriori informazioni su come Step Functions analizza un file delimitato da testo, vedere. Example of parsing an input CSV file

Quando l'esecuzione del workflow raggiunge lo Map stato, Step Functions richiama l'azione GetObjectAPI per recuperare il file specificato. Lo Map stato esegue quindi un'iterazione su ogni riga del file e avvia un'esecuzione secondaria del flusso di lavoro per elaborare gli elementi di ogni riga. Ad esempio, si supponga di fornire come input un file delimitato da testo contenente 100 righe. Quindi, l'interprete passa ogni riga allo stato. Map Lo Map stato elabora gli elementi in ordine seriale, a partire dalla riga di intestazione.

Le schede seguenti mostrano esempi della sintassi del ItemReader campo e dell'input passato all'esecuzione di un flusso di lavoro secondario per questo set di dati.

I file Parquet possono essere utilizzati come sorgente di input. I file Apache Parquet archiviati in Amazon S3 forniscono un'elaborazione dati colonnare efficiente su larga scala.

Quando si utilizzano file Parquet, si applicano le seguenti condizioni:

Di seguito viene mostrato un esempio di configurazione ASL da InputType impostare su Parquet:

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

È possibile utilizzare i file manifest Athena, generati dai risultati delle UNLOAD query, per specificare l'origine dei file di dati per lo stato della mappa. È stato ManifestType impostato suATHENA_DATA, e InputType su CSVJSONL, oParquet.

Durante l'esecuzione di una UNLOAD query, Athena genera un file manifesto di dati oltre agli oggetti dati effettivi. Il file manifest fornisce un elenco CSV strutturato dei file di dati. Sia il file manifest che i file di dati vengono salvati nella posizione dei risultati della query Athena in Amazon S3.

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

Panoramica concettuale del processo, in breve:

La funzionalità può elaborare i formati di output CSV, JSONL e Parquet di Athena. Tutti gli oggetti a cui si fa riferimento in un singolo file manifesto devono avere lo stesso formato. InputType Nota che gli oggetti CSV esportati da una UNLOAD query non includono l'intestazione nella prima riga. Verifica CSVHeaderLocation se devi fornire le intestazioni delle colonne.

Il contesto della mappa includerà anche un $states.context.Map.Item.Source modo per personalizzare l'elaborazione in base alla fonte dei dati.

Di seguito è riportato un esempio di configurazione di un file ItemReader configurato per l'utilizzo di un manifesto 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"
   }
}

Uno stato Distributed Map può accettare un file manifesto di inventario Amazon S3 archiviato in un bucket Amazon S3 come set di dati.

Quando l'esecuzione del flusso di lavoro raggiunge lo Map stato, Step Functions richiama l'azione GetObjectAPI per recuperare il file manifest di inventario Amazon S3 specificato.

Per impostazione predefinita, lo Map stato esegue quindi un'iterazione sugli oggetti nell'inventario per restituire una matrice di metadati degli oggetti di inventario Amazon S3.

Se si specifica ManifestType S3_INVENTORY, non può essere specificato. InputType

Di seguito è riportato un esempio di file di inventario in formato CSV. Questo file include gli oggetti denominati csvDataset andimageDataset, che sono archiviati in un bucket Amazon S3 denominato. 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"
...

Il seguente esempio di file manifesto di inventario di Amazon S3 mostra le intestazioni CSV per i metadati degli oggetti di inventario.

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

Le schede seguenti mostrano esempi della sintassi del ItemReader campo e dell'input passato all'esecuzione di un flusso di lavoro secondario per questo set di dati.