PySpark SQL：用作表达式的子查询返回的不止一行：答案

【问题标题】：PySpark SQL: more than one row returned by a subquery used as an expression:PySpark SQL：用作表达式的子查询返回的不止一行：
【发布时间】：2018-05-21 10:27:46
【问题描述】：

我试图使用来自另一个名为 train 的数据框的值在我的 test 数据框中创建一个新列。下面是一张快照。在train 数据框中，aml_freq_a 列下第一行的值是v 在a 列中出现的次数。同样，aml_freq_b 下的42 是l 在b 中出现的次数。 ['aml_freq_a', 'aml_freq_b', 'aml_freq_c'] 基本上是频率列。

>>> train.show(5)

+---+---+---+----------+----------+----------+                                  
|  a|  b|  c|aml_freq_a|aml_freq_b|aml_freq_c|
+---+---+---+----------+----------+----------+
|  v|  l|  l|        56|        42|        29|
|  u|  g|  l|        47|        46|        29|
|  s|  g|  l|        28|        46|        29|
|  v|  m|  l|        56|        33|        29|
|  h|  m|  l|        44|        33|        29|
+---+---+---+----------+----------+----------+

在test 数据集中有列['a', 'b', 'c']。在这里，我需要添加频率列 - ['aml_freq_a', 'aml_freq_b', 'aml_freq_c']。

>>> test.show(5)
+---+---+---+
|  a|  b|  c|
+---+---+---+
|  w|  j|  c|
|  a|  g|  w|
|  s|  d|  i|
|  g|  j|  r|
|  r|  b|  u|
+---+---+---+

为此，我编写了在a,b 和c 上加入train 和test 的子查询。

query = "select test.*,
  (select aml_freq_a from test left join train on test.a = train.a),
  (select aml_freq_b from test left join train on test.b = train.b),
  (select aml_freq_c from ten left join train on test.c = train.c)
from test"

train.createTempView('train')
test.createTempView('test')

spark.sql(query) 运行良好，但是当我在其上调用 show() 时，它会返回以下错误

java.lang.RuntimeException: more than one row returned by a subquery used as an expression:

这是什么意思？最初我认为我的查询有问题，但我验证了我的查询here，这里没有任何问题。我在这里没有看到什么？

【问题讨论】：

您可以编辑您的问题以向我们展示测试数据框吗？
@MichailN，我已经编辑包含它。
使用数据框连接并一个接一个地进行连接。测试数据框中的每一行都有多个连接行，因此不可能创建这样的数据框。 spark.apache.org/docs/2.1.0/api/python/…
还 spark.sql(query) “运行”很好，因为它还没有运行。 Spark 是惰性计算的，这意味着只有在触发动作时才会计算所有内容（如 show）请参阅：stackoverflow.com/questions/38027877/…
@MichailN 是的，是的。我知道惰性评估。我总是在我的所有数据帧上调用 show 方法。

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

【解决方案1】：

在您的尝试中，(select aml_freq_a from test left join train on test.a = train.a) 将返回一个dataframe，即多行，它不能用作"select test.*, (select aml_freq_a from test left join train on test.a = train.a),... 的选择参数

正确的查询如下

query = "select test.* from " \
            "(select test.*, aml_freq_a from " \
                "(select test.*, aml_freq_b from " \
                    "(select test.*, aml_freq_c from test " \
                "left join train on test.c = train.c) as test " \
            "left join train on test.b = train.b)  as test " \
        "left join train on test.a = train.a) as test"

如果需要以下格式的标题

+---+---+---+----------+----------+----------+
|a  |b  |c  |aml_freq_a|aml_freq_b|aml_freq_c|
+---+---+---+----------+----------+----------+

然后

query = "select test.* from " \
            "(select test.*, aml_freq_c from " \
                "(select test.*, aml_freq_b from " \
                    "(select test.*, aml_freq_a from test " \
                "left join train on test.a = train.a) as test " \
            "left join train on test.b = train.b)  as test " \
        "left join train on test.c = train.c) as test"

您可以通过更简单、更安全的方式使用 dataframe api

test.join(train.select('a', 'aml_freq_a'), ['a'], 'left') \
    .join(train.select('b', 'aml_freq_b'), ['b'], 'left') \
    .join(train.select('c', 'aml_freq_c'), ['c'], 'left')

希望回答对你有帮助

【讨论】：

这很有帮助。谢谢。
我必须在这里使用 sql，因为这部分进入一个函数，并且在我有三个以上变量的情况下，使用 sql 提供了更大的灵活性。除了易于理解之外，使用数据框 API 还有其他好处吗？另外，您提到了simpler and safer way。 safer 这里是什么意思？
我的意思是类型安全。 :)

【解决方案2】：

意思是

您使用的至少一个相关子查询返回多个匹配项。
虽然 Spark 仅支持为每一行返回一个值（即必须聚合相关的子查询）。

【讨论】：