Spark sql()是否支持谓词/投影下推？后续操作为何无法下推？

场景重现 #

先执行SQL语句：

val sql = s"""Select <column names> from <source> where <filter>"""
spark.sql(sql)

随后执行DataFrame操作：

val filteredDF = df.filter(filterExp)
// 或者
val selectDF = df.select(col1, col2, col3)

疑问 #

1. 谓词下推和投影下推优化是否会在.sql()命令中生效？
2. 后续执行.filter()或.select()操作时，这些操作能否下推至读取层？
3. 通过.explain()检查发现，情况(2)中的下推并未发生，这是为什么？

问题解答 #

1. .sql()中的下推优化是否生效？ #

会生效。Spark Catalyst优化器会对Spark SQL语句做全链路逻辑优化，包括谓词下推和投影下推——只要你的数据源支持下推（比如Parquet、ORC这类列式存储，或是JDBC数据源），Catalyst会自动把WHERE里的过滤条件推到数据源读取层，把SELECT指定的列限制也推过去，减少读取的数据量。

2. 后续的.filter()/.select()能否下推到读取层？ #

不一定，核心取决于操作的对象：

• 如果df是直接从数据源读取的原始DataFrame（比如spark.read.parquet(...)），后续的filter/select大概率能被下推；
• 但如果df是spark.sql(sql)执行后的结果，要看这个DataFrame是否保留了数据源的"可下推"元数据——如果之前的SQL已经触发了数据读取（比如执行过count()、show()这类action操作），或是优化器无法将后续操作与原始数据源的逻辑合并，下推就可能失效。

3. 为什么情况(2)的下推没发生？ #

常见原因有这几个：

• 数据已被物化：如果在spark.sql(sql)之后执行过count()、show()这类action，Spark会把SQL执行结果物化到内存/磁盘，后续的filter/select只能在物化后的数据集上操作，无法下推到原始数据源；
• 逻辑计划无法合并：Catalyst优化器的逻辑合并能力有限，如果之前的SQL包含复杂操作（比如聚合、窗口函数、自定义UDF），后续的filter/select无法和原始数据源的读取逻辑合并成可下推的计划；
• 数据源不支持：如果你的数据源本身不支持下推（比如普通文本文件、不支持下推的自定义数据源），不管是SQL还是后续DataFrame操作，都没法做下推；
• 执行计划被截断：spark.sql(sql)返回的DataFrame如果已经是优化后的物理计划，后续操作只能基于这个物理计划的结果继续处理，无法回溯到原始数据源做下推。

内容的提问来源于stack exchange，提问作者Kelly Jones