ItemReader (Map)

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 recorrerá en iteración 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, puede usar ItemsPath (solo Map, JSONPath) o una expresión de JSONata en el campo Items para los estados de JSONata.

Para procesar los elementos individuales, el estado Distributed Map 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 estado Map, Step Functions invoca la acción de la API ListObjectsV2, que devuelve una matriz de metadatos de objetos de Amazon S3. En esta matriz, cada elemento contiene datos, como ETag y Key, de los datos reales almacenados en el bucket.

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 API ListObjectsV2 contiene 100 elementos de metadatos. A continuación, el estado Distributed Map 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 ListObjectsV2 como origen de entrada en Distributed Map, sus máquinas de estado pueden leer y procesar varios objetos de datos directamente desde los buckets de Amazon S3, lo que elimina la necesidad de mapas anidados para procesar los metadatos.

Con la opción LOAD_AND_FLATTEN, su máquina de estado hará lo siguiente:

Con la opción de transformación, ya no necesitará Distributed Maps anidados para procesar los metadatos. El uso de la opción LOAD_AND_FLATTEN aumenta la escalabilidad, reduce los recuentos de ejecuciones de Map activo y procesa varios objetos de forma simultánea.

En la siguiente configuración se muestran los valores para 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 u objeto JSON.

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

Si el archivo JSON contiene una estructura de objetos anidados, puede seleccionar el nodo específico con su conjunto de datos con un ItemsPointer. Por ejemplo, la siguiente configuración extraería una lista anidada de 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 Distributed Map 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 llamado factcheck.jsonl. Ha almacenado este archivo en un prefijo llamado jsonlDataset en un bucket de Amazon S3. A continuación, se muestra un ejemplo del contenido de un 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 Map Distributed puede aceptar un archivo delimitado por texto almacenado en un bucket de Amazon S3 como conjunto de datos. Si usa 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, consulte Contenido del campo ItemReader.

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

Para obtener más información acerca de 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 estado Map, Step Functions invoca la acción de la API GetObject para recuperar el archivo especificado. A continuación, el estado Map recorre en iteración cada fila del archivo e inicia la ejecución de un flujo de trabajo secundario para procesar los elementos de cada fila. Por ejemplo, suponga 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 de Parquet se pueden utilizar como origen de entrada. Los archivos de Apache Parquet almacenados en Amazon S3 proporcionan un procesamiento de datos en columnas eficiente y a escala.

Cuando se utilicen archivos de Parquet, se aplicarán las siguientes condiciones:

A continuación, se muestra un ejemplo de configuración de ASL para InputType establecido en 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 la consulta UNLOAD, para especificar el origen de los archivos de datos de su estado Map. Se establece ManifestType en ATHENA_DATA, y InputType en CSV, JSONL o Parquet.

Al ejecutar una consulta UNLOAD, 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. Los archivos de datos y de manifiesto se guardan en la ubicación de resultados de consulta de Athena en Amazon S3.

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

Descripción conceptual del proceso, en resumen:

La característica 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 de InputType. Tenga en cuenta que los objetos CSV exportados mediante una consulta UNLOAD no incluyen el encabezado en la primera línea. Compruebe CSVHeaderLocation si tiene que proporcionar los encabezados de columnas.

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

A continuación se muestra un ejemplo de configuración de un ItemReader configurado 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 estado Map, Step Functions invoca la acción de la API GetObject para recuperar el archivo de manifiesto de inventario de Amazon S3 especificado.

De forma predeterminada, el estado Map 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_INVENTORY, no se puede especificar InputType.

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 de los objetos 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.