Spark SQL 中的相关子查询答案

【问题标题】：Correlated Subquery in Spark SQLSpark SQL 中的相关子查询
【发布时间】：2020-07-25 07:20:37
【问题描述】：

我有以下 2 个表，我必须使用 correlated 子查询检查它们之间是否存在值。

要求是——对于orders表中的每条记录，检查customer表中是否存在对应的custid，然后输出一个值为Y的字段（名为FLAG），如果custid 存在，否则 N 如果不存在。

订单：

orderid | custid
12345   | XYZ
34566   | XYZ
68790   | MNP
59876   | QRS
15620   | UVW

客户：

id | custid
1  | XYZ
2  | UVW

预期输出：

orderid | custid  | FLAG
12345   | XYZ     | Y
34566   | XYZ     | Y 
68790   | MNP     | N
59876   | QRS     | N
15620   | UVW     | Y

我尝试了以下类似的方法，但无法正常工作 -

select 
o.orderid,
o.custid,
case when o.custid EXISTS (select 1 from customer c on c.custid = o.custid)
     then 'Y'
     else 'N'
end as flag
from orders o

这可以通过 correlated scalar 子查询来解决吗？如果不是，实现此要求的最佳方式是什么？

请指教。

注意：使用 Spark SQL 查询 v2.4.0

谢谢。

【问题讨论】：

感谢您支持我的回答。我现在已经测试过它可以与本地重新创建的数据副本一起使用。如果它对你有用，你会接受我的回答吗？我和你一样对 StackOverflow 比较陌生，每一点都很重要！谢谢！
@Lars Skaug 感谢您的回复。您的解决方案效果很好，我已经接受了答案。但是，只是想知道是否可以对EXISTS 子句/correlated subquery 做同样的事情（据我所知）？
也许在未来。例如，您可以在 Oracle 中执行此操作。当你在 spark 2.4 中尝试它时，你实际上会得到一个信息性错误，说“IN/EXISTS 谓词子查询只能在 Spark 中的过滤器中使用。”内容丰富，尽管这不是您所希望的。

标签： apache-spark apache-spark-sql

【解决方案1】：

IN/EXISTS 谓词子查询只能在 Spark 中的过滤器中使用。

以下内容适用于本地重新创建的数据副本：

select orderid, custid, case when existing_customer is null then 'N' else 'Y' end existing_customer
          from (select o.orderid, o.custid, c.custid existing_customer
                from orders o
                left join customer c
                 on c.custid = o.custid)

这是它如何处理重新创建的数据：

def textToView(csv: String, viewName: String) = {
   spark.read
  .option("ignoreLeadingWhiteSpace", "true")
  .option("ignoreTrailingWhiteSpace", "true")
  .option("delimiter", "|")
  .option("header", "true")
  .csv(spark.sparkContext.parallelize(csv.split("\n")).toDS)
  .createOrReplaceTempView(viewName)
}

textToView("""id | custid
              1  | XYZ
              2  | UVW""", "customer")

textToView("""orderid | custid
              12345   | XYZ
              34566   | XYZ
              68790   | MNP
              59876   | QRS
              15620   | UVW""", "orders")

spark.sql("""
          select orderid, custid, case when existing_customer is null then 'N' else 'Y' end existing_customer
          from (select o.orderid, o.custid, c.custid existing_customer
                from orders o
                left join customer c
                 on c.custid = o.custid)""").show

+-------+------+-----------------+
|orderid|custid|existing_customer|
+-------+------+-----------------+
|  59876|   QRS|                N|
|  12345|   XYZ|                Y|
|  34566|   XYZ|                Y|
|  68790|   MNP|                N|
|  15620|   UVW|                Y|
+-------+------+-----------------+

【讨论】：