PySpark SQL: 用作表达式的子查询返回的多行：

Question

我尝试使用另一个名为 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() 它时 returns 我出现以下错误

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

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

Answer 1

表示

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

Answer 2

在您的尝试中，(select aml_freq_a from test left join train on test.a = train.a) 会 return 一个 dataframe 即多行不能用作 select 参数为 "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"

如果 header 需要以下格式

+---+---+---+----------+----------+----------+
|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"

你可以用更简单、更安全的方式做到这一点使用数据框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')

希望回答对你有帮助

PySpark SQL: 用作表达式的子查询返回的多行：

PySpark SQL: more than one row returned by a subquery used as an expression:

apache-spark

apache-spark-sql

pyspark

pyspark-sql