ItemReader (Karte)

Ein Distributed-Map-Status kann eine JSON-Eingabe akzeptieren, die aus einem vorherigen Schritt im Workflow übergeben wurde.

Die Eingabe kann ein JSON-Array, ein JSON-Objekt oder ein Array innerhalb eines Knotens eines JSON-Objekts sein.

Step Functions iteriert direkt über die Elemente eines Arrays oder die Schlüssel-Wert-Paare eines JSON-Objekts.

Um einen bestimmten Knoten auszuwählen, der ein Array aus einem JSON-Objekt enthält, können Sie den ItemsPath ( JSONPath Nur Karte) oder einen JSONata Ausdruck im Items Feld für Bundesstaaten verwenden. JSONata

Um einzelne Elemente zu verarbeiten, startet der Status Distributed Map für jedes Element eine untergeordnete Workflow-Ausführung. Die folgenden Registerkarten zeigen Beispiele für die an den Map Status übergebenen Eingaben und die entsprechenden Eingaben für die Ausführung eines untergeordneten Workflows.

Ein Distributed Map-Status kann über die Objekte iterieren, die in einem Amazon S3 S3-Bucket gespeichert sind. Wenn die Workflow-Ausführung den Map Status erreicht, ruft Step Functions die ListObjectsV2-API-Aktion auf, die ein Array der Amazon S3 S3-Objektmetadaten zurückgibt. In diesem Array enthält jedes Element Daten, wie z. B. ETagund Key, für die tatsächlichen Daten, die im Bucket gespeichert sind.

Um einzelne Elemente im Array zu verarbeiten, startet der Status Distributed Map die Ausführung eines untergeordneten Workflows. Nehmen wir zum Beispiel an, dass Ihr Amazon S3 S3-Bucket 100 Bilder enthält. Dann enthält das nach dem Aufrufen der ListObjectsV2 API-Aktion zurückgegebene Array 100 Metadatenelemente. Im Status Distributed Map werden dann 100 untergeordnete Workflow-Ausführungen gestartet, um jedes Element zu verarbeiten.

Um Datenobjekte direkt und ohne verschachtelte Workflows zu verarbeiten, können Sie die Transformationsoption LOAD_AND_FLATTEN wählen, um Elemente direkt zu verarbeiten.

Die folgenden Registerkarten zeigen Beispiele für die ItemReader Feldsyntax und die Eingabe, die an die Ausführung eines untergeordneten Workflows für diesen Datensatz übergeben wurde.

Dank der verbesserten Unterstützung für S3 ListObjects V2 als Eingabequelle in Distributed Map können Ihre Zustandsmaschinen mehrere Datenobjekte direkt aus Amazon S3 S3-Buckets lesen und verarbeiten, sodass keine verschachtelten Maps zur Verarbeitung der Metadaten erforderlich sind!

Mit dieser LOAD_AND_FLATTEN Option macht Ihr State Machine Folgendes:

Mit der Transformationsoption benötigen Sie keine verschachtelten verteilten Maps mehr, um die Metadaten zu verarbeiten. Die Verwendung der Option LOAD_AND_FLATTEN erhöht die Skalierbarkeit, reduziert die Anzahl der aktiven Kartenläufe und verarbeitet mehrere Objekte gleichzeitig.

Die folgende Konfiguration zeigt die Einstellung für: ItemReader

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

Ein Distributed Map-Status kann eine JSON-Datei, die in einem Amazon S3 S3-Bucket gespeichert ist, als Datensatz akzeptieren. Die JSON-Datei muss ein Array oder ein JSON-Objekt enthalten.

Wenn die Workflow-Ausführung den Map Status erreicht, ruft Step Functions die GetObjectAPI-Aktion auf, um die angegebene JSON-Datei abzurufen.

Wenn die JSON-Datei eine verschachtelte Objektstruktur enthält, können Sie den spezifischen Knoten mit Ihrem Datensatz mit einem auswählen. ItemsPointer Die folgende Konfiguration würde beispielsweise eine verschachtelte Liste von ausgewählten Produkten im Inventar extrahieren.

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

Der Map Status wiederholt dann jedes Element im Array und startet für jedes Element eine untergeordnete Workflow-Ausführung. Wenn Ihre JSON-Datei beispielsweise 1000 Array-Elemente enthält, startet der Map Status 1000 untergeordnete Workflow-Ausführungen.

Die folgenden Registerkarten zeigen Beispiele für die ItemReader Feldsyntax und die Eingabe, die an die Ausführung eines untergeordneten Workflows für diesen Datensatz übergeben wurde.

Stellen Sie sich für dieses Beispiel vor, Sie haben eine JSON-Datei mit dem Namenfactcheck.json. Sie haben diese Datei in einem Präfix gespeichert, das jsonDataset in einem Amazon S3 S3-Bucket benannt ist. Das Folgende ist ein Beispiel für den JSON-Datensatz.

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

Ein Distributed Map-Status kann eine JSON Lines-Datei, die in einem Amazon S3 S3-Bucket gespeichert ist, als Datensatz akzeptieren.

Die folgenden Registerkarten zeigen Beispiele für die ItemReader Feldsyntax und die Eingabe, die an die Ausführung eines untergeordneten Workflows für diesen Datensatz übergeben wurde.

Stellen Sie sich für dieses Beispiel vor, Sie haben eine JSON Lines-Datei mit dem Namenfactcheck.jsonl. Sie haben diese Datei in einem Präfix gespeichert, das jsonlDataset in einem Amazon S3 S3-Bucket benannt ist. Das Folgende ist ein Beispiel für den Inhalt der Datei.

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

Ein Distributed Map-Status kann eine durch Text getrennte Datei, die in einem Amazon S3 S3-Bucket gespeichert ist, als Datensatz akzeptieren. Wenn Sie eine durch Text getrennte Datei als Datensatz verwenden, müssen Sie eine Spaltenüberschrift angeben. Hinweise zum Angeben einer Kopfzeile finden Sie unter. Inhalt des ItemReader Felds

Step Functions analysiert textgetrennte Dateien auf der Grundlage der folgenden Regeln:

Weitere Informationen darüber, wie Step Functions eine durch Text getrennte Datei analysiert, finden Sie unter. Example of parsing an input CSV file

Wenn die Workflow-Ausführung den Map Status erreicht, ruft Step Functions die GetObjectAPI-Aktion auf, um die angegebene Datei abzurufen. Der Map Status iteriert dann über jede Zeile in der Datei und startet eine untergeordnete Workflow-Ausführung, um die Elemente in jeder Zeile zu verarbeiten. Nehmen wir beispielsweise an, Sie stellen eine durch Text getrennte Datei bereit, die 100 Zeilen als Eingabe enthält. Dann übergibt der Interpreter jede Zeile an den Status. Map Der Map Status verarbeitet Elemente in serieller Reihenfolge, beginnend nach der Kopfzeile.

Die folgenden Registerkarten zeigen Beispiele für die ItemReader Feldsyntax und die Eingabe, die an die Ausführung eines untergeordneten Workflows für diesen Datensatz übergeben wurde.

Parquet-Dateien können als Eingabequelle verwendet werden. Apache Parquet-Dateien, die in Amazon S3 gespeichert sind, ermöglichen eine effiziente, skalierbare, spaltenweise Datenverarbeitung.

Bei der Verwendung von Parquet-Dateien gelten die folgenden Bedingungen:

Im Folgenden finden Sie ein Beispiel für eine ASL-Konfiguration für die InputType Einstellung „Parquet“:

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

Sie können die aus den UNLOAD Abfrageergebnissen generierten Athena-Manifestdateien verwenden, um die Quelle der Datendateien für Ihren Kartenstatus anzugeben. Sie legen fest ManifestType InputType auf ATHENA_DATA und entweder CSVJSONL, oderParquet.

Beim Ausführen einer UNLOAD Abfrage generiert Athena zusätzlich zu den eigentlichen Datenobjekten eine Datenmanifestdatei. Die Manifestdatei enthält eine strukturierte CSV-Liste der Datendateien. Sowohl das Manifest als auch die Datendateien werden an Ihrem Speicherort für Athena-Abfrageergebnisse in Amazon S3 gespeichert.

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

Konzeptioneller Überblick über den Prozess, kurz gefasst:

Die Funktion kann die Ausgabeformate CSV, JSONL und Parquet von Athena verarbeiten. Alle Objekte, auf die in einer einzigen Manifestdatei verwiesen wird, müssen dasselbe Format haben. InputType Beachten Sie, dass durch eine UNLOAD Abfrage exportierte CSV-Objekte keinen Header in der ersten Zeile enthalten. Prüfen SieCSVHeaderLocation, ob Sie Spaltenüberschriften angeben müssen.

Der Kartenkontext wird auch eine enthalten, $states.context.Map.Item.Source sodass Sie die Verarbeitung an die Datenquelle anpassen können.

Im Folgenden finden Sie eine Beispielkonfiguration eines für die Verwendung eines Athena-Manifests ItemReader konfigurierten:

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

Ein Distributed Map-Status kann eine Amazon S3 S3-Inventarmanifestdatei, die in einem Amazon S3 S3-Bucket gespeichert ist, als Datensatz akzeptieren.

Wenn die Workflow-Ausführung den Map Status erreicht, ruft Step Functions die GetObjectAPI-Aktion auf, um die angegebene Amazon S3 S3-Inventarmanifestdatei abzurufen.

Standardmäßig iteriert der Map Status dann über die Objekte im Inventar, um ein Array von Amazon S3 S3-Inventarobjekt-Metadaten zurückzugeben.

Wenn Sie S3_INVENTORY angeben, kann ManifestType es InputType nicht angegeben werden.

Im Folgenden finden Sie ein Beispiel für eine Inventardatei im CSV-Format. Diese Datei enthält die Objekte mit den Namen csvDataset undimageDataset, die in einem benannten Amazon S3 S3-Bucket gespeichert sindamzn-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"
...

Das folgende Beispiel für eine Amazon S3 S3-Inventarmanifestdatei zeigt die CSV-Header für die Metadaten des Inventarobjekts.

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

Die folgenden Registerkarten zeigen Beispiele für die ItemReader Feldsyntax und die Eingabe, die an die Ausführung eines untergeordneten Workflows für diesen Datensatz übergeben wurde.