除非您自己实现了DataSource,否则不能直接使用谓词下推。谓词下推是由Spark数据源提供的机制,每个数据源必须单独实现。
对于基于文件的数据源,已经有一个简单的基于磁盘分区的机制。
考虑以下DataFrame:
val df = Seq(("test", "day1"), ("test2", "day2")).toDF("data", "day")
如果我们以以下方式将该DataFrame保存到磁盘上:
df.write.partitionBy("day").save("/tmp/data")
结果将是以下文件夹结构
tmp -
|
| - data - |
|
|--day=day1 -|- part1....parquet
| |- part2....parquet
|
|--day=day2 -|- part1....parquet
|- part2....parquet
如果您现在是这样使用此数据源的:
spark.read.load("/tmp/data").filter($"day" = "day1").show()
Spark不会加载day2文件夹的数据,因为没有必要。
这是一种谓词下推的类型,适用于spark支持的每种标准文件格式。
更具体的机制是parquet。Parquet是一种基于列的文件格式,这意味着很容易过滤掉列。如果您有基于parquet的文件,其中包含3列a
、b
、c
,在文件/tmp/myparquet.parquet
中,以下查询:
spark.read.parquet("/tmp/myparquet.parquet").select("a").show()
这将导致内部谓词下推,其中Spark仅获取列a
的数据,而不读取列b
或c
的数据。
如果有人对实现此特性所建立的机制感兴趣:
@Stable
trait PrunedFilteredScan {
def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row]
}
可以在https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala中找到。