我的团队正在使用Spark构建ETL过程，将原始分隔符文本文件加载到基于Parquet的“数据湖”中。 Parquet列存储的一个承诺是查询只会读取必要的“列条纹”。
但是，对于嵌套模式结构，我们发现正在阅读意外的列。
为了说明这一点，这里有一个使用Scala和Spark 2.0.1 shell的POC：
// 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")
然后，我们将文件读取回DataFrame，并投影到列的子集：
// 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
当运行此命令时，我们会看到预期的输出：
+---+-------+
| F1|Orig_F1|
+---+-------+
| 1| 1|
| 3| 3|
+---+-------+
但是...查询计划显示了略微不同的情况：
“优化后的计划”显示：
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
“解释”即表示：
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>>
执行期间生成的 INFO 日志也确认了 Orig.F2 列被意外读取：
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))
根据Dremel论文和Parquet文档，复杂嵌套结构的列应该独立存储和独立检索。
问题：
- 这种行为是当前Spark查询引擎的限制吗？换句话说，Parquet是否支持最优执行此查询，但Spark的查询计划者很简单？
- 还是说这是当前Parquet实现的限制？
- 还是我没有正确使用Spark API？
- 或者，我误解了Dremel / Parquet列存储的工作原理？