spark 谓词下推是否适用于 JDBC？答案

【问题标题】：Does spark predicate pushdown work with JDBC?spark 谓词下推是否适用于 JDBC？
【发布时间】：2015-12-10 23:56:02
【问题描述】：

根据this

Catalyst 应用逻辑优化，例如谓词下推。这优化器可以将过滤谓词下推到数据源中，使物理执行能够跳过不相关的数据。

Spark 支持将谓词下推到数据源。此功能是否也适用于/预期适用于 JDBC？

（通过检查数据库日志，我可以看到这不是现在的默认行为 - 完整的查询被传递到数据库，即使它后来受到 spark 过滤器的限制）

更多详情

使用 PostgreSQL 9.4 运行 Spark 1.5

代码sn-p：

from pyspark import SQLContext, SparkContext, Row, SparkConf
from data_access.data_access_db import REMOTE_CONNECTION

sc = SparkContext()
sqlContext = SQLContext(sc)

url = 'jdbc:postgresql://{host}/{database}?user={user}&password={password}'.format(**REMOTE_CONNECTION)
sql = "dummy"

df = sqlContext.read.jdbc(url=url, table=sql)
df = df.limit(1)
df.show()

SQL 跟踪：

< 2015-09-15 07:11:37.718 EDT >LOG:  execute <unnamed>: SET extra_float_digits = 3                                                                                                                      
< 2015-09-15 07:11:37.771 EDT >LOG:  execute <unnamed>: SELECT * FROM dummy WHERE 1=0                                                                                                                   
< 2015-09-15 07:11:37.830 EDT >LOG:  execute <unnamed>: SELECT c.oid, a.attnum, a.attname, c.relname, n.nspname, a.attnotnull OR (t.typtype = 'd' AND t.typnotnull), pg_catalog.pg_get_expr(d.adbin, d.a
drelid) LIKE '%nextval(%' FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON (c.relnamespace = n.oid) JOIN pg_catalog.pg_attribute a ON (c.oid = a.attrelid) JOIN pg_catalog.pg_type t ON (a.a
tttypid = t.oid) LEFT JOIN pg_catalog.pg_attrdef d ON (d.adrelid = a.attrelid AND d.adnum = a.attnum) JOIN (SELECT 15218474 AS oid , 1 AS attnum UNION ALL SELECT 15218474, 3) vals ON (c.oid = vals.oid
 AND a.attnum = vals.attnum)                                                                                                                                                                            
< 2015-09-15 07:11:40.936 EDT >LOG:  execute <unnamed>: SET extra_float_digits = 3                                                                                                                      
< 2015-09-15 07:11:40.964 EDT >LOG:  execute <unnamed>: SELECT "id","name" FROM dummy

我希望最后一个选择将包含 limit 1 子句 - 但它没有

【问题讨论】：

只需在 SQL 代码本身中添加限制（或其他过滤器）
您是否已将 where 子句传递给 postgres 的推送过滤器？

标签： python jdbc apache-spark apache-spark-sql pyspark

【解决方案1】：

Spark DataFrames 支持 JDBC 源的谓词下推，但术语 predicate 用于严格的 SQL 含义。这意味着它仅涵盖WHERE 子句。此外，它看起来仅限于逻辑连词（恐怕没有IN 和OR）和简单的谓词。

限制、计数、排序、组和条件等其他所有内容都在 Spark 端进行处理。 SO 中已经提到的一个警告是，df.count() 或 sqlContext.sql("SELECT COUNT(*) FROM df") 被转换为 SELECT 1 FROM df，并且需要使用 Spark 进行大量数据传输和处理。

这是否意味着这是一个失败的事业？不完全是。可以使用任意子查询作为table 参数。它不如谓词下推方便，但效果很好：

n = ... # Number of rows to take
sql = "(SELECT * FROM dummy LIMIT {0}) AS tmp".format(int(n))
df = sqlContext.read.jdbc(url=url, table=sql)

注意：

一旦数据源 API v2 准备就绪，这种行为可能会在未来得到改进：

【讨论】：