0

Puntualmente estoy buscando optimizar el rendimiento al actualizar e insertar datos a una tabla base DeltaLake, con cerca de 4 billones de registros.

Entorno : Spark 3.0.0 DeltaLake 0.7.0

En contexto esto trata sobre realizar una tabla incremental por medio de DeltaLake, resumiré esto en pasos para ser más detallado :

  1. Creación de la tabla base (delta)
  2. Obtención de datos periódicos
  3. Agregar los datos a la tabla base

Los pasos 1 y 2 ya están realizados , pero al momento de agregar los datos el rendimiento es notoriamente lento , por ejemplo agregar un csv de 9GB demora cerca de 6 hora, esto principalmente porque delta necesita reescribir los datos por cada actualización , además necesita "leer" todos los datos de la base.

Esta tabla además está particionada (PARTITIONED BY) y almacenada en GDFS (HDFS) del cluster para asegurar que los nodos spark puedan realizar las operaciones.

Los campos de la tabla base (cardinalidad asignada por # ):

  • ID : Identificador , #10000
  • TIPO : string , #30
  • FECHA_LOCAL : Fecha local del registro
  • FECHA_UTC : Fecha utc de registro
  • VALUE : Valor del registro
  • YEAR : Columna calculada int #4
  • MONTH : columna calculada int # 12
  • DAY : columna calculada int # 31

Como la búsqueda general es por tiempo , se decidió particionar por la columna FECHA_LOCAL en YEAR , MONTH, DAY , se descartó particionar por las columnas ID y FECHA_LOCAL por su alto nivel de cardinalidad , (que para efectos de rendimiento es peor), se agregó finalmente TIPO , quedando dela siguiente manera:

spark.sql(f"""
CREATE OR REPLACE TABLE  {TABLE_NAME} (
  ID INT, 
  FECHA_LOCAL TIMESTAMP,
  FECHA_UTC TIMESTAMP,
  TIPO STRING, 
  VALUE DOUBLE,
  YEAR INT,
  MONTH INT, 
  DAY INT )
USING DELTA
PARTITIONED BY (YEAR , MONTH , DAY, TIPO)
LOCATION '{location}'
""")

A partir de ahora la incrementalidad viene dada por agregar periódicamente estos archivos csv de app 9Gb cada 5 días.

Actualmente la operación MERGE queda de la siguiente manera :

# Se lee la base ya creada  
spark.sql(f"DROP TABLE IF EXISTS {BASE_TABLE_NAME}")
spark.sql(f"""CREATE TABLE {BASE_TABLE_NAME} USING DELTA LOCATION '{base_path}'""")

# Se lee el incremetal  ya creado
spark.sql(f"DROP TABLE IF EXISTS {INCREMENTAL_TABLE_NAME}") 
spark.sql(f"""CREATE TABLE {INCREMENTAL_TABLE_NAME} USING DELTA LOCATION '{inc_path}'""")


spark.sql(f"""
        MERGE INTO {BASE_TABLE_NAME}
        USING {INCREMENTAL_TABLE_NAME} ON 
            --partitioned cols
            {BASE_TABLE_NAME}.YEAR  = {INCREMENTAL_TABLE_NAME}.YEAR AND
            {BASE_TABLE_NAME}.MONTH = {INCREMENTAL_TABLE_NAME}.MONTH AND  
            {BASE_TABLE_NAME}.DAY   = {INCREMENTAL_TABLE_NAME}.DAY AND 
            {BASE_TABLE_NAME}.TIPO = {INCREMENTAL_TABLE_NAME}.TIPO AND 


            {BASE_TABLE_NAME}.FECHA_LOCAL= {INCREMENTAL_TABLE_NAME}.FECHA_LOCALAND 
            {BASE_TABLE_NAME}.ID= {INCREMENTAL_TABLE_NAME}.ID

        WHEN MATCHED  THEN  
        UPDATE SET    {BASE_TABLE_NAME}.VALUE= {INCREMENTAL_TABLE_NAME}.VALUE,
                      {BASE_TABLE_NAME}.TIPO= {INCREMENTAL_TABLE_NAME}.TIPO


        WHEN NOT MATCHED THEN
        INSERT *

        """)

Algunos datos para considerar:

  • El tiempo de esta operación MERGE de de 6 horas
  • La tabla base se creó a partir de 230GB datos csv (55gb ahora en delta!)
  • La configuración de al aplicación spark es en modo cluster con los siguientes parámetros
  • La infra consta de 3 nodos, de 32 núcleos y 250GB RAM c/u , aunque se ocupa menos por seguridad de las demás aplicaciones existentes app -50% de recursos.

Spark app

mode = 'spark://spark-master:7077'
#mode='local[*]'
spark = (SparkSession.builder.master(mode)
        .appName(" SparkApp ") 
        .config('spark.cores.max', '45')
        .config('spark.executor.cores', '5') 
        .config('spark.executor.memory', '11g')   
        .config('spark.driver.memory', '120g')   
        .config("spark.sql.shuffle.partitions", f"200") #200 only for 
        lecturas delta table 200GB
        .config("spark.storage.memoryFraction", f"0.8")
        # DeltaLake configs
        .config("spark.jars.packages", "io.delta:delta-core_2.12:0.7.0")
        .config("spark.sql.extensions", 
        "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", 
        "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        # Delta optimization 
        .config("spark.databricks.delta.optimizeWrite.enabled", "true")
        .config("spark.databricks.delta.retentionDurationCheck.enabled", 
        "false")
        .getOrCreate()
)

1 respuesta 1

0

Bueno opto por compartir esta respuesta para que puedan aprovechar algunos consejos.

Delta recomienda utilizar todas la columnas particionadas, de este modo la búsqueda final de datos es menor, dado por el efecto de "poda"

Así que es necesario identificar todos los casos en donde el merge puede actualizar los datos, para ello se realiza una consulta a los datos incrementales para generar un diccionario de este tipo:

filter_columns = spark.sql(f""" 
SELECT 
    YEAR,
    MONTH, 
    DAY,
    COLLECT_LIST( DISTINCT  TIPO) AS TIPOS
FROM incremental 
GROUP BY YEAR , MONTH  , DAY 
ORDER BY 1 , 2 , 3
""").toPandas()

Con este df es posible se generar las condiciones en donde el merge debe actualizar/insertar:

df agrupado por año , mes , día , tipo

Luego generó un string llamado "final_cond" de esta forma :

dic = filter_columns.groupby(['YEAR','MONTH','DAY'])['TIPO'].apply(lambda grp: list(grp.value_counts().index)).to_dict()
final_cond=''
index=0
for key , value in dic.items(): 
    cond=''
    year=key[0]
    month=key[1]
    day=key[2] 
    variables = ','.join(["'"+str(x)+"'" for x in value[0]])
    or_cond= '' if  index+1 == len(dic) else '\nOR\n'
    
    cond =  f"""({BASE_TABLE_NAME}.YEAR == {year} AND {BASE_TABLE_NAME}.MONTH == {month} AND {BASE_TABLE_NAME}.DAY =={day} AND {BASE_TABLE_NAME}.TIPO IN ({variables}) )"""
      
    final_cond = final_cond+ cond + f'{or_cond}'
    index+=1
    #break
    
print(final_cond)

introducir la descripción de la imagen aquí

Finalmente agregamos estas condiciones al MERGE :

...
WHEN MATCHED AND ({final_cond}) THEN
...

Este simple "filtro" redujo el tiempo del merge de 6 horas a 49 minutos !!!

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.