Necesito ayuda para entender el flujo de ejecución de Spark SQL para una condición dada.
Tenemos un código spark escrito en scala:
val sqlContent = "Consulta compleja en hive..."
val df = spark.sql(sqlContent)
df.coalesce(3).write.option("orc.compress", "snappy").mode("sobrescribir").insertInto(hivetable)
Aquí está el comando para enviar este código:
spark-submit --executor-memory 8G --num-executors 8 --executor-cores 4 --driver-memory 8G --queue ETL --conf spark.sql.sources.partitionOverwriteMode=dynamic --conf spark.sql.shuffle.partitions=100 --jars brickhouse-0.7.1-SNAPSHOT.jar --conf spark.executor.extraLibraryPath=/usr/hdp/current/hadoop-client/lib/native:/usr/hdp/current/hadoop-client/lib/native/Linux-amd64-64:/usr/lib/x86_64-linux-gnu --class com.abc.MySparkProgram --master yarn --deploy-mode cluster hdfs://user/com/abc/MySparkProject.jar <algunos parámetros de entrada>
Cuando se ejecuta este trabajo, capturé una captura de pantalla desde Spark-SQL yarn ui:
Aquí están mis preguntas:
1- Ayúdame a entender el flujo que se menciona en la captura de pantalla anterior.
2- Como he establecido coalesce=3, lo que significa que todas las particiones/archivos se fusionarán en 3 particiones/archivos. Pero si miras el último rectángulo de la captura de pantalla anterior (es decir, el rectángulo Execute InsertIntoHiveTable), está mostrando: "número de archivos escritos: 234" así como "número de partes dinámicas: 234". ¿Por qué? ¿Y este trabajo tardó 1 hora y 30 minutos en finalizar?
3- Cada vez que envío este trabajo con los mismos parámetros de entrada, cada vez me muestra un "número de archivos escritos: 44" así como "número de partes dinámicas: 44" diferentes en la misma "Ejecutar rectángulo InsertIntoHiveTable". ¿Por qué? ¿Y este trabajo tardó 25 minutos en finalizar?
davy.ai
¿Por favor me ayudas a entender el flujo que se menciona en la captura de pantalla anterior?
El flujo en la captura de pantalla es el siguiente:
1. El código Spark se ejecuta, que lee una consulta compleja de Hive y almacena los resultados en un DataFrame.
2. El DataFrame se fusiona en 3 particiones/archivos.
3. El DataFrame se escribe en una tabla de Hive usando el comando
insertInto, con laopcióndeorc.compressestablecida ensnappy.4. Luego se envía el trabajo al clúster YARN, con los parámetros de configuración especificados.
5. Luego se ejecuta el trabajo y los resultados se escriben en la tabla de Hive.
Como configuré coalesce=3, lo que significa que todas las particiones/archivos se fusionarán en 3 particiones/archivos. Pero si observa el último rectángulo de la captura de pantalla anterior (es decir, el rectángulo Ejecutar InsertIntoHiveTable), muestra: “número de archivos escritos: 234” y “número de partes dinámicas: 234”. ¿Por qué? Y este trabajo tardó 1 hora y 30 minutos en terminar.
El comando
coalescese utiliza para reducir el número de particiones/archivos en el DataFrame. Sin embargo, cuando el DataFrame se escribe en la tabla de Hive, el número de particiones/archivos puede aumentar debido a la particionamiento dinámico de los datos. Es por eso que el número de archivos escritos es 234, aunque el DataFrame se fusionó en 3 particiones/archivos. El trabajo tardó 1 hora y 30 minutos en finalizar debido al particionamiento dinámico de los datos, que requiere más tiempo para procesar.Cada vez que presento este trabajo con los mismos parámetros de entrada, siempre me muestra diferentes “número de archivos escritos: 44” y “número de partes dinámicas: 44” en el mismo “Ejecutar InsertIntoHiveTable rectángulo”. ¿Por qué? Y este trabajo tardó 25 minutos en finalizar.
El número de archivos escritos y particiones dinámicas puede ser diferente cada vez que se presenta el trabajo, incluso con los mismos parámetros de entrada, porque los datos pueden estar particionados de manera diferente cada vez. Es por eso que el número de archivos escritos y las particiones dinámicas pueden ser diferentes cada vez que el trabajo se presenta. El trabajo tardó 25 minutos en finalizar porque los datos se particionaron de manera más eficiente esta vez, lo que resultó en un tiempo de ejecución más rápido.