Optimización de las lecturas con inserción en operaciones de ETL de AWS Glue

El método Pushdown es una técnica de optimización que acerca la lógica de la recuperación de datos al origen de los datos. El origen podría ser una base de datos o un sistema de archivos como Amazon S3. Al ejecutar determinadas operaciones directamente en el origen, puede ahorrar tiempo y potencia de procesamiento al no llevar todos los datos de la red al motor Spark gestionado por Glue AWS.

Otra forma de decir esto es que la presión hacia abajo reduce el escaneo de datos. Para obtener más información sobre el proceso de identificación de cuándo esta técnica es adecuada, consulte Reducir la cantidad de datos escaneados en la guía de Prácticas recomendadas para el ajuste del rendimiento de AWS Glue para trabajos de Apache Spark en las Recomendaciones de AWS.

Predica cómo insertar los archivos almacenados en Amazon S3

Al trabajar con archivos en Amazon S3 organizados por prefijo, puede filtrar las rutas de Amazon S3 de destino mediante la definición de un predicado de inserción. En lugar de leer el conjunto de datos completo y aplicar filtros dentro de un DynamicFrame, puede aplicar el filtro directamente a los metadatos de la partición almacenados en el catálogo de datos de Glue AWS. Este enfoque permite enumerar y leer de forma selectiva solo los datos necesarios. Para obtener más información sobre este proceso, incluida la escritura en un bucket mediante particiones, consulte Administración de particiones para la salida de ETL en AWS Glue.

Para lograr la inserción de predicados en Amazon S3, utilice el parámetro push_down_predicate. Pensemos en un bucket de Amazon S3 que haya particionado por año, mes y día. Si quiere recuperar los datos de los clientes de junio de 2022, puede indicar a Glue AWS que lea solo las rutas relevantes de Amazon S3. En este caso, el push_down_predicate es year='2022' and month='06'. Juntándolo todo, la operación de lectura se puede lograr de la siguiente manera:

Python

customer_records = glueContext.create_dynamic_frame.from_catalog( 
    database = "customer_db", 
    table_name = "customer_tbl",
    push_down_predicate = "year='2022' and month='06'"
)

Scala

val customer_records = glueContext.getCatalogSource(
database="customer_db", 
tableName="customer_tbl", 
pushDownPredicate="year='2022' and month='06'"
).getDynamicFrame()

En el escenario anterior, push_down_predicate recupera una lista de todas las particiones del catálogo de datos de Glue AWS y las filtra antes de leer los archivos subyacentes de Amazon S3. Aunque esto ayuda en la mayoría de los casos, cuando se trabaja con conjuntos de datos que tienen millones de particiones, el proceso de enumerar las particiones puede llevar mucho tiempo. Para solucionar este problema, se puede utilizar la depuración de las particiones desde el servidor para mejorar el rendimiento. Para ello, se crea un índice de particiones para los datos en el catálogo de datos de Glue AWS. Para obtener más información sobre los índices de particiones, consulte Creación de índices de particiones . A continuación, puede utilizar la opción catalogPartitionPredicate para hacer referencia al índice. Para ver un ejemplo de cómo recuperar particiones con catalogPartitionPredicate, consulte Filtrado del lado del servidor mediante predicados de partición de catálogo.

Insertar al trabajar con fuentes JDBC

El lector JDBC de Glue AWS que se utiliza en el archivo GlueContext admite la inserción en las bases de datos compatibles al proporcionar consultas SQL personalizadas que se pueden ejecutar directamente en el origen. Esto se puede lograr mediante la configuración del parámetro sampleQuery. La consulta de ejemplo puede especificar qué columnas seleccionar, así como proporcionar un predicado de inserción para limitar los datos transferidos al motor Spark.

De forma predeterminada, las consultas de muestra funcionan en un único nodo, lo que puede provocar fallos en el trabajo cuando se manejan grandes volúmenes de datos. Para utilizar esta función para consultar datos a escala, debe configurar la partición de consultas con el valor enablePartitioningForSampleQuery a verdadero, lo que distribuirá la consulta a varios nodos en la clave que elija. La partición de consultas también requiere algunos otros parámetros de configuración necesarios. Para obtener más información sobre la partición de consultas, vea Lectura desde tablas de JDBC en paralelo.

Al configurar enablePartitioningForSampleQuery, Glue AWS combinará el predicado de inserción con un predicado de partición al consultar la base de datos. Su sampleQuery debe terminar con un AND para que Glue AWS agregue condiciones de partición. (Si no proporciona un predicado de inserción, sampleQuery debe terminar con un WHERE). Consulte un ejemplo a continuación, en el que presionamos un predicado hacia abajo para recuperar solo las id filas superiores a 1000. Esto sampleQuery solo devolverá las columnas de nombre y ubicación de las filas donde id sea mayor que el valor especificado:

Python

sample_query = "select name, location from customer_tbl WHERE id>=1000 AND"
customer_records = glueContext.create_dynamic_frame.from_catalog(
    database="customer_db",
    table_name="customer_tbl",
    sample_query = "select name, location from customer_tbl WHERE id>=1000 AND",

    additional_options = { 
                           "hashpartitions": 36 , 
                           "hashfield":"id",
                           "enablePartitioningForSampleQuery":True, 
                           "sampleQuery":sample_query
                          }
)

Scala

val additionalOptions = Map( 
        "hashpartitions" -> "36", 
        "hashfield" -> "id", 
        "enablePartitioningForSampleQuery" -> "true", 
        "sampleQuery" -> "select name, location from customer_tbl WHERE id >= 1000 AND"
        )
 
    val customer_records = glueContext.getCatalogSource(
        database="customer_db", 
        tableName="customer_tbl").getDynamicFrame()
                    

nota

Si el nombre de customer_tbl es diferente en el Catálogo de datos en el almacén de datos subyacente, debe brindar el nombre subyacente de la tabla en sample_query, ya que la solicitud es enviada al almacén de datos subyacente.

También puede realizar consultas en tablas JDBC sin necesidad de integrarlas con el catálogo de datos de Glue AWS. En lugar de proporcionar el nombre de usuario y la contraseña como parámetros del método, puede reutilizar las credenciales de una conexión preexistente y proporcionar useConnectionProperties y connectionName. En este ejemplo, recuperamos las credenciales de una conexión llamada my_postgre_connection.

Python

connection_options_dict = {
    "useConnectionProperties": True,
    "connectionName": "my_postgre_connection",
    "dbtable":"customer_tbl",
    "sampleQuery":"select name, location from customer_tbl WHERE id>=1000 AND",
    "enablePartitioningForSampleQuery":True,
    "hashfield":"id",
    "hashpartitions":36
    }

customer_records = glueContext.create_dynamic_frame.from_options(
    connection_type="postgresql",
    connection_options=connection_options_dict
    )

Scala

val connectionOptionsJson = """
      {
        "useConnectionProperties": true,
        "connectionName": "my_postgre_connection",
        "dbtable": "customer_tbl",
        "sampleQuery": "select name, location from customer_tbl WHERE id>=1000 AND",
        "enablePartitioningForSampleQuery" : true,
        "hashfield" : "id",
        "hashpartitions" : 36
      }
    """
    
    val connectionOptions = new JsonOptions(connectionOptionsJson)
    
    val dyf = glueContext.getSource("postgresql", connectionOptions).getDynamicFrame()

Notas y limitaciones de cómo insertar en Glue AWS

El concepto Pushdown es aplicable cuando se lee desde orígenes que no son de streaming. AWS Glue admite una variedad de orígenes: la capacidad de insertar depende de la fuente y el conector.

• Cuando se conecte a Snowflake, puede usar la opción query. Existe una funcionalidad similar en el conector Redshift de Glue 4.0 AWS y versiones posteriores. Para obtener más información acerca de cómo leer de Snowflake withquery, consulte Lectura de tablas de Snowflake.

• El lector de ETL de DynamoDB no es compatible con filtros ni predicados de inserción. MongoDB y DocumentDB tampoco admiten este tipo de funcionalidad.

• Al leer datos almacenados en Amazon S3 en formatos de tabla abierta, el método de partición de los archivos en Amazon S3 ya no es suficiente. Para leer y escribir desde particiones que utilizan formatos de tabla abierta, consulte la documentación del formato.

• Los métodos DynamicFrame no realizan la inserción de proyección de Amazon S3. Se leerán todas las columnas de los archivos que pasen el filtro de predicados.

• Cuando se trabaja con conectores custom.jdbc en Glue AWS, la capacidad de inserción depende del origen y del conector. Revise la documentación del conector correspondiente para confirmar si es compatible con la inserción en Glue AWS y cómo lo hace.