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 :
- Creación de la tabla base (delta)
- Obtención de datos periódicos
- 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()
)