Dlaczego Apache Spark odczytuje niepotrzebne kolumny parkietu w strukturach zagnieżdżonych?

Question

Mój zespół buduje proces ETL, który ładuje nieprzetworzone pliki tekstowe do "jeziora danych" opartego na parkiecie za pomocą Sparka. Jedną z obietnic magazynu kolumn Parquet jest to, że zapytanie odczyta tylko niezbędne "pasy kolumn".

Widzimy jednak nieoczekiwane kolumny odczytane dla zagnieżdżonych struktur schematu.

Aby wykazać, tutaj jest POC użyciu Scala i iskra 2.0.1 Obudowa:

// 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") 

Następnie czytamy plik z powrotem do DataFrame i projektu do podzbioru kolumn:

// 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 

Gdy ta biegnie widzimy oczekiwany wynik:

+---+-------+ 
| F1|Orig_F1| 
+---+-------+ 
| 1|  1| 
| 3|  3| 
+---+-------+ 

ale ... plan kwerend pokazuje nieco tarasowy t historia:

z "zoptymalizowane planu" pokazuje:

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 

i "wyjaśnić" pokazuje:

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>> 

I dzienniki INFO powstałe podczas realizacji również potwierdzić, że kolumna jest Orig.F2 niespodziewanie przeczytać:

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)) 

Według Dremel paper i Parquet documentation, kolumny dla złożonych struktur zagnieżdżonych powinny być niezależnie przechowywane i niezależnie odzyskiwane.

Pytania:

1. Czy to zachowanie ograniczenie obecnego silnika zapytań Spark? Innymi słowy, czy Parquet wspiera optymalnie wykonywanie tego zapytania, ale program do planowania zapytań Sparka jest naiwny?
2. Czy jest to ograniczenie obecnej wersji Parkietu?
3. Czy nie używam poprawnie interfejsów API Spark?
4. Czy nie rozumiem, w jaki sposób powinno działać przechowywanie kolumn Dremel/Parquet?

prawdopodobnie związane: Why does the query performance differ with nested columns in Spark SQL?

Answer 1

Jest to ograniczenie na silniku zapytań Spark w tej chwili, odpowiedni bilet JIRA jest poniżej, iskra obsługuje tylko źródłowe kreację prostych typów w parkiet, nie zagnieżdżone StructTypes

https://issues.apache.org/jira/browse/SPARK-17636