pyspark-sql - pyspark，在 WHERE 之外不支持引用外部查询

Question

我需要在 pyspark 中加入 2 个表，并且不根据右表中的精确值执行此联接，而是根据最接近的值(因为没有精确匹配。

它在常规 SQL 中工作正常，但在 SparkSQL 中不起作用。
我正在使用 Spark 2.2.1

在常规 SQL 中:

SELECT a.*,
(SELECT b.field2 FROM tableB b 
WHERE b.field1 = a.field1 
ORDER BY ABS(b.field2 - a.field2) LIMIT 1) as field2
FROM tableA a
ORDER BY a.field1

工作正常

在 SparkSQL 中:

...
tableA_DF.registerTempTable("tableA")
tableB_DF.registerTempTable("tableB")

query = "SELECT a.*, \
(SELECT b.field2 FROM tableB b \
WHERE b.field1 = a.field1 \
ORDER BY ABS(b.field2 - a.field2) LIMIT 1) field2 \
FROM tableA a \
ORDER BY a.field1"

result_DF = spark.sql(query) 

我有以下异常(exception):

pyspark.sql.utils.AnalysisException:在 WHERE/HAVING 子句之外不支持引用外部查询的 u'Expressions

如果 Spark 2.2.1 不支持它，那么解决方法是什么？

先感谢您，
加里

Answer 1

最好的办法是重构查询，使 field2 值从连接表而不是标量相关子查询输出。例如:

...
tableA_DF.registerTempTable("tableA")
tableB_DF.registerTempTable("tableB")

query = "SELECT a.*, \
FIRST(b.field2) OVER (ORDER BY ABS(b.field2 - a.field2)) field2 \
FROM tableA a \
JOIN tableB b
ON a.field1 = b.field1 \
ORDER BY a.field1"

result_DF = spark.sql(query) 

当以这种方式编写时，Catalyst 能够解析查询计划。请注意，上述内容在 Spark 2.3.1 上进行了测试 - 可能需要一些最近引入的窗口功能才能使其工作。

对于可能无法重新定义 JOIN 或 WHERE 子句的人，Spark 2.4 可能包含一些用于相关子查询处理的新功能:
https://issues.apache.org/jira/browse/SPARK-18455

更新:不幸的是，SPARK-18455 中提到的相关子查询在 2018 年 9 月 11 日滑到了目标版本 3.0.0。在这一点上，我们不太可能看到包含该功能的 2.x 版本，而且从历史上看，为基本主要版本规范的 Spark 次要功能已从路线图上滑落。目前，我认为相关子查询主要不在 Spark 团队的路线图中。