代码之家 › 专栏 › 技术社区 › Matthias

分组表达式上的相关子查询-TreeNodeException:Binding属性,tree:count(1)#382L

pyspark-sql apache-spark-sql pyspark apache-spark

0

Matthias · 技术社区 · 7 年前

spark.createDataFrame([
    Row(a=5, b=10), Row(a=5, b=10), Row(a=5, b=10),
    Row(a=6, b=10), Row(a=6, b=10), Row(a=6, b=10), Row(a=6, b=10), Row(a=6, b=10), Row(a=6, b=10),
    Row(a=5, b=11), Row(a=5, b=11),
    Row(a=6, b=12), Row(a=6, b=12), Row(a=6, b=12), Row(a=6, b=12),
    Row(a=5, b=5), Row(a=5, b=5), Row(a=5, b=5), Row(a=5, b=5), Row(a=5, b=5), Row(a=5, b=5), Row(a=5, b=5),
]).registerTempTable('mydata')

首先,我只是简单地计算每对存在的频率:

spark.sql('''
SELECT a, b,
    COUNT(*) as count
FROM mydata AS o
GROUP BY a, b
''').show()

输出:

+---+---+-----+
|  a|  b|count|
+---+---+-----+
|  6| 12|    4|
|  5|  5|    7|
|  6| 10|    6|
|  5| 10|    3|
|  5| 11|    2|
+---+---+-----+

现在,我想添加一个额外的列,其中包含一对存在的频率与a中具有相同值的对总数的百分比。为此,我尝试添加一个计算总数的相关子查询:

spark.sql('''
SELECT a, b,
    COUNT(*) as count,
    (COUNT(*) / (
        SELECT COUNT(*) FROM mydata AS i WHERE o.a = i.a
    )) as percentage
FROM mydata AS o
GROUP BY a, b
''').show()

我期待的是:

+---+---+-----+----------+
|  a|  b|count|percentage|
+---+---+-----+----------+
|  6| 12|    4|       0.4|  --> 10 pairs exist with a=6 --> 4/10 = 0.4
|  5|  5|    7|    0.5833|  --> 12 pairs exist with a=5 --> 7/12  =0.5833
|  6| 10|    6|       0.6|  --> ...
|  5| 10|    3|      0.25|
|  5| 11|    2|    0.1666|
+---+---+-----+----------+

py4j.protocol.Py4JJavaError: An error occurred while calling o371.showString.
: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Binding attribute, tree: count(1)#382L
    at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56)
    at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:91)
    at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:90)
[...]
Caused by: java.lang.RuntimeException: Couldn't find count(1)#382L in [a#305L,b#306L,count(1)#379L]
    at scala.sys.package$.error(package.scala:27)
    at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1$$anonfun$applyOrElse$1.apply(BoundAttribute.scala:97)
    at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1$$anonfun$applyOrElse$1.apply(BoundAttribute.scala:91)
    at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:52)
    ... 80 more

这听起来有点混乱-Pypark不知怎么想访问内部连接的计数?

1 回复 | 直到 7 年前

1

akuiper 7 年前

从第一个表中,您可以使用窗口函数计算百分比; sum(count) over (partition by a) 将计算 count 通过 a

spark.sql('''
    SELECT a, b,
        COUNT(*) as count
    FROM mydata AS o
    GROUP BY a, b
''').registerTempTable('count')

spark.sql('''
    SELECT *, 
           count / sum(count) over (partition by a) as percentage 
    FROM count
''').show()
+---+---+-----+-------------------+
|  a|  b|count|         percentage|
+---+---+-----+-------------------+
|  6| 12|    4|                0.4|
|  6| 10|    6|                0.6|
|  5|  5|    7| 0.5833333333333334|
|  5| 10|    3|               0.25|
|  5| 11|    2|0.16666666666666666|
+---+---+-----+-------------------+