ItemReader (Mapa)

Un estado Map Distributed puede aceptar una entrada JSON transferida desde un paso anterior del flujo de trabajo.

La entrada puede ser una matriz JSON, un objeto JSON o una matriz dentro de un nodo de un objeto JSON.

Step Functions iterará directamente sobre los elementos de una matriz o los pares clave-valor de un objeto JSON.

Para seleccionar un nodo específico que contenga una matriz de un objeto JSON, puedes usar la expresión ItemsPath ( JSONPath Solo en el mapa) o usar una JSONata expresión en el Items campo para los estados. JSONata

Para procesar elementos individuales, el estado del mapa distribuido inicia la ejecución de un flujo de trabajo secundario para cada elemento. Las siguientes pestañas muestran ejemplos de la entrada que se transfiere al estado Map y la entrada correspondiente a la ejecución de un flujo de trabajo secundario.

Un estado Map Distributed puede iterar sobre los objetos que se almacenan en un bucket de Amazon S3. Cuando la ejecución del flujo de trabajo alcanza el Map estado, Step Functions invoca la acción de la API ListObjectsV2, que devuelve una matriz de metadatos del objeto Amazon S3. En esta matriz, cada elemento contiene datos, como ETaguna clave, de los datos reales almacenados en el depósito.

Para procesar los elementos individuales de la matriz, el estado Map Distributed inicia la ejecución de un flujo de trabajo secundario. Por ejemplo, suponga que su bucket de Amazon S3 contiene 100 imágenes. A continuación, la matriz devuelta tras invocar la acción de la ListObjectsV2 API contiene 100 elementos de metadatos. A continuación, el estado del mapa distribuido inicia 100 ejecuciones de flujos de trabajo secundarios para procesar cada elemento.

Para procesar los objetos de datos directamente, sin flujos de trabajo anidados, puede elegir la opción de transformación LOAD_AND_FLATTEN para procesar los elementos directamente.

En las siguientes pestañas se muestran ejemplos de la sintaxis del campo ItemReader y de la entrada que se transfiere a la ejecución de un flujo de trabajo secundario para este conjunto de datos.

Gracias a la compatibilidad mejorada con S3 ListObjects V2 como fuente de entrada en Distributed Map, sus máquinas de estado pueden leer y procesar varios objetos de datos directamente de los buckets de Amazon S3, ¡lo que elimina la necesidad de mapas anidados para procesar los metadatos!

Con LOAD_AND_FLATTEN esta opción, su máquina de estados hará lo siguiente:

Con la opción de transformación, ya no necesitará mapas distribuidos anidados para procesar los metadatos. El uso de la opción LOAD_AND_FLATTEN aumenta la escalabilidad, reduce el número de reproducciones de mapas activos y procesa varios objetos de forma simultánea.

La siguiente configuración muestra la configuración de 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"
   }
}

Un estado Map Distributed puede aceptar un archivo JSON almacenado en un bucket de Amazon S3 como conjunto de datos. El archivo JSON debe contener una matriz o un objeto JSON.

Cuando la ejecución del flujo de trabajo alcanza el Map estado, Step Functions invoca la acción de la GetObjectAPI para recuperar el archivo JSON especificado.

Si el archivo JSON contiene una estructura de objetos anidados, puedes seleccionar el nodo específico con tu conjunto de datos con una. ItemsPointer Por ejemplo, la siguiente configuración extraería una lista anidada de los productos destacados del 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"
   }
}

A continuación, el estado Map se repite sobre cada elemento de la matriz e inicia la ejecución de un flujo de trabajo secundario para cada elemento. Por ejemplo, si el archivo JSON contiene 1000 elementos de matriz, el estado Map inicia 1000 ejecuciones de flujos de trabajo secundarios.

En las siguientes pestañas se muestran ejemplos de la sintaxis del campo ItemReader y de la entrada que se transfiere a la ejecución de un flujo de trabajo secundario para este conjunto de datos.

Para este ejemplo, imagine que tiene un archivo JSON llamado factcheck.json. Ha almacenado este archivo en un prefijo llamado jsonDataset en un bucket de Amazon S3. A continuación se muestra un ejemplo de conjunto de datos 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"
  },
  ...
]

Un estado de mapa distribuido puede aceptar un archivo de líneas JSON almacenado en un bucket de Amazon S3 como conjunto de datos.

En las siguientes pestañas se muestran ejemplos de la sintaxis del campo ItemReader y de la entrada que se transfiere a la ejecución de un flujo de trabajo secundario para este conjunto de datos.

Para este ejemplo, imagine que tiene un archivo de líneas JSON llamadofactcheck.jsonl. Ha almacenado este archivo en un prefijo llamado jsonlDataset en un bucket de Amazon S3. El siguiente es un ejemplo del contenido del archivo.

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

Un estado de mapa distribuido puede aceptar un archivo delimitado por texto que esté almacenado en un bucket de Amazon S3 como conjunto de datos. Si utiliza un archivo delimitado por texto como conjunto de datos, debe especificar un encabezado de columna. Para obtener información sobre cómo especificar un encabezado, consulteContenido del ItemReader campo.

Step Functions analiza los archivos delimitados por texto según las siguientes reglas:

Para obtener más información sobre cómo Step Functions analiza un archivo delimitado por texto, consulte. Example of parsing an input CSV file

Cuando la ejecución del flujo de trabajo alcanza el Map estado, Step Functions invoca la acción de la GetObjectAPI para recuperar el archivo especificado. A continuación, el Map estado se repite en cada fila del archivo e inicia la ejecución de un flujo de trabajo secundario para procesar los elementos de cada fila. Por ejemplo, supongamos que proporciona un archivo delimitado por texto que contiene 100 filas como entrada. Entonces, el intérprete pasa cada fila al estado Map. El estado Map procesa los elementos en orden de serie, a partir de la fila siguiente al encabezado.

En las siguientes pestañas se muestran ejemplos de la sintaxis del campo ItemReader y de la entrada que se transfiere a la ejecución de un flujo de trabajo secundario para este conjunto de datos.

Los archivos Parquet se pueden utilizar como fuente de entrada. Los archivos de Apache Parquet almacenados en Amazon S3 proporcionan un procesamiento de datos en columnas eficiente a escala.

Al utilizar archivos Parquet, se aplican las siguientes condiciones:

A continuación se muestra un ejemplo de configuración de ASL para InputType configurar Parquet:

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

Puede usar los archivos de manifiesto de Athena, generados a partir de los resultados de las UNLOAD consultas, para especificar la fuente de los archivos de datos del estado de su mapa. Se configura ManifestType comoATHENA_DATA, y InputType como CSVJSONL, oParquet.

Al ejecutar una UNLOAD consulta, Athena genera un archivo de manifiesto de datos además de los objetos de datos reales. El archivo de manifiesto proporciona una lista CSV estructurada de los archivos de datos. Tanto el manifiesto como los archivos de datos se guardan en la ubicación de resultados de la consulta de Athena en Amazon S3.

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

Resumen conceptual del proceso:

La función puede procesar los formatos de salida CSV, JSONL y Parquet de Athena. Todos los objetos a los que se hace referencia en un único archivo de manifiesto deben tener el mismo formato. InputType Tenga en cuenta que los objetos CSV exportados mediante una UNLOAD consulta no incluyen el encabezado en la primera línea. Comprueba CSVHeaderLocation si necesitas proporcionar los encabezados de las columnas.

El contexto del mapa también incluirá una opción $states.context.Map.Item.Source para que pueda personalizar el procesamiento en función de la fuente de los datos.

El siguiente es un ejemplo de configuración de una ItemReader configuración para usar un manifiesto de 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"
   }
}

Un estado Map Distributed puede aceptar un archivo de manifiesto de inventario de Amazon S3 almacenado en un bucket de Amazon S3 como conjunto de datos.

Cuando la ejecución del flujo de trabajo alcanza el Map estado, Step Functions invoca la acción de la GetObjectAPI para recuperar el archivo de manifiesto de inventario de Amazon S3 especificado.

De forma predeterminada, el Map estado recorre en iteración los objetos del inventario para devolver una matriz de metadatos de objetos de inventario de Amazon S3.

Si especifica que ManifestType es S3_INVESTORY, no se InputType puede especificar.

A continuación se muestra un ejemplo de un archivo de inventario de ejemplo en formato CSV: Este archivo incluye los objetos denominados csvDataset yimageDataset, que se almacenan en un bucket de Amazon S3 que lleva ese nombre 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"
...

El siguiente ejemplo de un archivo de manifiesto de inventario de Amazon S3 muestra los encabezados CSV de los metadatos del objeto de 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"
  } ]
}

En las siguientes pestañas se muestran ejemplos de la sintaxis del campo ItemReader y de la entrada que se transfiere a la ejecución de un flujo de trabajo secundario para este conjunto de datos.