Reducción de la cantidad de análisis de datos

Tamaño del archivo: recomendamos mantener el tamaño de los archivos de entrada y salida dentro de un rango moderado (por ejemplo, 128 MB). Los archivos demasiado pequeños y demasiado grandes pueden causar problemas.

Un gran número de archivos pequeños provoca los siguientes problemas:

Por otro lado, si el tipo de archivo no se puede dividir (por ejemplo, gzip) y los archivos son demasiado grandes, la aplicación de Spark debe esperar a que una sola tarea termine de leer todo el archivo.

Para reducir el paralelismo excesivo que se produce cuando se crea una tarea de Apache Spark para cada archivo pequeño, utilice la agrupación de archivos para DynamicFrames. Este enfoque reduce las probabilidades de que se produzca una excepción de OOM por parte del controlador de Spark. Para configurar la agrupación de archivos, defina los parámetros groupFiles y groupSize. En el siguiente ejemplo de código se usa la API DynamicFrame de AWS Glue en un script de ETL con estos parámetros.

dyf = glueContext.create_dynamic_frame_from_options("s3",
    {'paths': ["s3://input-s3-path/"],
    'recurse':True,
    'groupFiles': 'inPartition',
    'groupSize': '1048576'},
    format="json")

Compresión: si los objetos de S3 ocupan cientos de megabytes, considere la posibilidad de comprimirlos. Hay varios formatos de compresión, que se pueden clasificar a grandes rasgos en dos tipos:

En el caso de Spark (y otros motores de procesamiento distribuido habituales), dividirá el archivo de datos de origen en fragmentos que su motor pueda procesar en paralelo. A estas unidades se las suele denominar divisiones. Una vez que los datos estén en un formato que se pueda dividir, los lectores de AWS Glue optimizados pueden recuperar las divisiones de un objeto de S3 al ofrecer la opción Range a la API GetObject para recuperar solo bloques específicos. Tenga en cuenta el siguiente diagrama para ver cómo funcionaría esto en la práctica.

Los datos comprimidos pueden acelerar considerablemente la aplicación, siempre que los archivos tengan un tamaño óptimo o se puedan dividir. Los tamaños de datos más pequeños reducen los datos analizados de Amazon S3 y el tráfico de red de Amazon S3 a su clúster de Spark. Por otro lado, se necesita más CPU para comprimir y descomprimir los datos. La cantidad de computación necesaria escala con la relación de compresión del algoritmo de compresión. Tenga en cuenta esta concesión a la hora de elegir el formato de compresión que se pueda dividir.

Formato de archivo: utilice un formato de columnas. Apache Parquet y Apache ORC son formatos de datos de columnas habituales. Parquet y ORC almacenan los datos de manera eficiente mediante la compresión basada en columnas, codificando y comprimiendo cada columna en función de su tipo de datos. Para obtener más información sobre las codificaciones de Parquet, consulte Parquet encoding definitions. Los archivos de Parquet también se pueden dividir.

Los formatos de columnas agrupan los valores por columnas y los almacenan juntos en bloques. Cuando utilice formatos de columnas, puede omitir bloques de datos que correspondan a columnas que no tiene previsto utilizar. Las aplicaciones de Spark pueden recuperar solo las columnas que necesita. En general, unos índices de compresión mejores o la omisión de bloques de datos implican leer menos bytes de Amazon S3, lo que se traduce en un mejor rendimiento. Ambos formatos también admiten los siguientes enfoques de inserción para reducir la E/S:

Diseño de archivo: al almacenar los datos de S3 en objetos situados en diferentes rutas en función de cómo se vayan a utilizar los datos, podrá recuperar los datos pertinentes de forma eficiente. Para obtener más información, consulte Organizar objetos con prefijos en la documentación de Amazon S3. AWS Glue admite almacenar claves y valores en prefijos de Amazon S3 en formato key=value y los datos se particionan según la ruta de Amazon S3. La partición de los datos le permite restringir el volumen de datos que analiza cada aplicación de análisis posterior, lo que mejora el rendimiento y reduce los costos. Para obtener más información, consulte Administración de particiones para la salida de ETL en AWS Glue.

La partición divide la tabla en diferentes partes y mantiene los datos relacionados en archivos agrupados en función de los valores de las columnas, como el año, el mes y el día, como se muestra en el siguiente ejemplo.

# Partitioning by /YYYY/MM/DD
s3://<YourBucket>/year=2023/month=03/day=31/0000.gz
s3://<YourBucket>/year=2023/month=03/day=01/0000.gz
s3://<YourBucket>/year=2023/month=03/day=02/0000.gz
s3://<YourBucket>/year=2023/month=03/day=03/0000.gz
...

Si quiere definir particiones para su conjunto de datos, modélelo con una tabla en el AWS Glue Data Catalog. A continuación, puede restringir la cantidad de datos analizados mediante la eliminación de particiones, de la siguiente manera:

Demasiadas particiones de Amazon S3: evite particionar los datos de Amazon S3 en columnas que contengan una amplia gama de valores, como una columna de ID con miles de valores. Esto puede aumentar considerablemente el número de particiones del bucket, ya que el número de particiones posibles es el producto de todos los campos por los que ha creado particiones. Demasiadas particiones pueden provocar lo siguiente:

Por ejemplo, si establece un tipo de fecha en partitionBy o partitionKeys, la creación de particiones de nivel de fecha, como yyyy/mm/dd, es adecuada para muchos casos de uso. Sin embargo, yyyy/mm/dd/<ID> podría generar tantas particiones que afectaría negativamente al rendimiento en su conjunto.

Por otro lado, algunos casos de uso, como las aplicaciones de procesamiento en tiempo real, requieren muchas particiones, como yyyy/mm/dd/hh. Si su caso de uso requiere muchas particiones, considere la posibilidad de utilizar índices de particiones de AWS Glue para reducir la latencia a la hora de recuperar los metadatos de las particiones del Catálogo de datos.

Para DynamicFrame, utilice la opción sampleQuery. Al utilizar create_dynamic_frame.from_catalog, configure el argumento additional_options de la siguiente manera.

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_catalog(
  database = db,
  table_name = table,
  additional_options={
      "sampleQuery": query,
      "hashexpression": key,
      "hashpartitions": 10,
      "enablePartitioningForSampleQuery": True
    },
  transformation_ctx = "datasource0"
)

Cuando using create_dynamic_frame.from_options, configure el argumento connection_options de la siguiente manera.

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type = connection, 
    connection_options={
        "url": url,
        "user": user,
        "password": password,
        "dbtable": table,
        "sampleQuery": query,
        "hashexpression": key,
        "hashpartitions": 10,
        "enablePartitioningForSampleQuery": True
    }
)

Para DataFrame, utilice la opción query.

query = "SELECT * FROM <TableName> where id = 'XX'"
jdbcDF = spark.read \
.format('jdbc') \
.option('url', url) \
.option('user', user) \
.option('password', pwd) \
.option('query', query) \
.load()

En el caso de Amazon Redshift, utilice la versión 4.0 o posteriores de AWS Glue para aprovechar la compatibilidad con la inserción en el conector de Amazon Redshift Spark.

dyf = glueContext.create_dynamic_frame.from_catalog(
    database = "redshift-dc-database-name", 
    table_name = "redshift-table-name", 
    redshift_tmp_dir = args["temp-s3-dir"], 
    additional_options = {"aws_iam_role": "arn:aws:iam::role-account-id:role/rs-role-name"}
)

Para otras bases de datos, consulte la documentación correspondiente.

Para evitar un análisis completo de todas las ejecuciones continuas de trabajos y procesar solo los datos que no estuvieron presentes durante la última ejecución del trabajo, habilite los marcadores de trabajos.

Para limitar la cantidad de datos de entrada que se van a procesar, habilite la ejecución delimitada con marcadores de trabajos. Esto ayuda a reducir la cantidad de datos analizados para cada ejecución de trabajo.