sacar - ¿Por qué Apache Spark lee columnas de parquet innecesarias dentro de estructuras anidadas?
programacion ats estructuras (1)
Mi equipo está creando un proceso ETL para cargar archivos de texto delimitados en bruto en un "lago de datos" basado en Parquet usando Spark. Una de las promesas del almacén de columnas de Parquet es que una consulta solo leerá las "franjas de columnas" necesarias.
Pero estamos viendo columnas inesperadas que se leen para estructuras de esquema anidadas.
Para demostrarlo, aquí hay un POC que usa Scala y el shell Spark 2.0.1:
// Preliminary setup
sc.setLogLevel("INFO")
import org.apache.spark.sql.types._
import org.apache.spark.sql._
// Create a schema with nested complex structures
val schema = StructType(Seq(
StructField("F1", IntegerType),
StructField("F2", IntegerType),
StructField("Orig", StructType(Seq(
StructField("F1", StringType),
StructField("F2", StringType))))))
// Create some sample data
val data = spark.createDataFrame(
sc.parallelize(Seq(
Row(1, 2, Row("1", "2")),
Row(3, null, Row("3", "ABC")))),
schema)
// Save it
data.write.mode(SaveMode.Overwrite).parquet("data.parquet")
Luego leemos el archivo nuevamente en un DataFrame y lo proyectamos en un subconjunto de columnas:
// Read it back into another DataFrame
val df = spark.read.parquet("data.parquet")
// Select & show a subset of the columns
df.select($"F1", $"Orig.F1").show
Cuando esto se ejecuta vemos el resultado esperado:
+---+-------+
| F1|Orig_F1|
+---+-------+
| 1| 1|
| 3| 3|
+---+-------+
Pero ... el plan de consulta muestra una historia ligeramente diferente:
El "plan optimizado" muestra:
val projected = df.select($"F1", $"Orig.F1".as("Orig_F1"))
projected.queryExecution.optimizedPlan
// Project [F1#18, Orig#20.F1 AS Orig_F1#116]
// +- Relation[F1#18,F2#19,Orig#20] parquet
Y "explicar" muestra:
projected.explain
// == Physical Plan ==
// *Project [F1#18, Orig#20.F1 AS Orig_F1#116]
// +- *Scan parquet [F1#18,Orig#20] Format: ParquetFormat, InputPaths: hdfs://sandbox.hortonworks.com:8020/user/stephenp/data.parquet, PartitionFilters: [], PushedFilters: [], ReadSchema: struct<F1:int,Orig:struct<F1:string,F2:string>>
Y los registros INFO producidos durante la ejecución también confirman que la columna Orig.F2 se lee inesperadamente:
16/10/21 15:13:15 INFO parquet.ParquetReadSupport: Going to read the following fields from the Parquet file:
Parquet form:
message spark_schema {
optional int32 F1;
optional group Orig {
optional binary F1 (UTF8);
optional binary F2 (UTF8);
}
}
Catalyst form:
StructType(StructField(F1,IntegerType,true), StructField(Orig,StructType(StructField(F1,StringType,true), StructField(F2,StringType,true)),true))
De acuerdo con el documento de Dremel y la documentación de Parquet , las columnas para estructuras anidadas complejas deben almacenarse independientemente y ser recuperables independientemente.
Preguntas:
- ¿Es este comportamiento una limitación del motor de consulta Spark actual? En otras palabras, ¿soporta Parquet la ejecución óptima de esta consulta, pero el planificador de consultas de Spark es ingenuo?
- O, ¿es esto una limitación de la implementación actual de Parquet?
- O, ¿no estoy usando las API de Spark correctamente?
- O, ¿estoy entendiendo mal cómo se supone que debe funcionar el almacenamiento de columnas Dremel / Parquet?
Posiblemente relacionado: ¿Por qué el rendimiento de la consulta difiere con las columnas anidadas en Spark SQL?
Es una limitación en el motor de consulta de Spark en este momento, el boleto JIRA relevante está debajo, la chispa solo maneja el empuje predicado de tipos simples en Parquet, no en StructTypes anidados