pyspark 根据条件计算行数答案

【问题标题】：pyspark count rows on conditionpyspark 根据条件计算行数
【发布时间】：2018-08-07 21:14:25
【问题描述】：

我有一个数据框

test = spark.createDataFrame([('bn', 12452, 221), ('mb', 14521, 330),('bn',2,220),('mb',14520,331)],['x','y','z'])
test.show()

我需要根据条件计算行数：

test.groupBy("x").agg(count(col("y")>12453),count(col("z")>230)).show()

这给了

 +---+------------------+----------------+
 |  x|count((y > 12453))|count((z > 230))|
 +---+------------------+----------------+
 | bn|                 2|               2|
 | mb|                 2|               2|
 +---+------------------+----------------+

这只是行数，而不是某些条件下的行数。

【问题讨论】：

标签： count pyspark

【解决方案1】：

count 不求和 Trues，它只计算非空值的数量。要计算 True 值，需要先将条件转换为 1 / 0 然后sum：

import pyspark.sql.functions as F

cnt_cond = lambda cond: F.sum(F.when(cond, 1).otherwise(0))
test.groupBy('x').agg(
    cnt_cond(F.col('y') > 12453).alias('y_cnt'), 
    cnt_cond(F.col('z') > 230).alias('z_cnt')
).show()
+---+-----+-----+
|  x|y_cnt|z_cnt|
+---+-----+-----+
| bn|    0|    0|
| mb|    2|    2|
+---+-----+-----+

【讨论】：

从显示表中，有没有办法可以将值提取到 Python 变量？ *.com/questions/53689509/…
我可以在这里检查一下我对 pyspark 的理解吗：这里的 lambda 函数都在 spark 中，所以这永远不必创建用户定义的 python 函数，并伴随着相关的减速。正确的？这看起来很方便。
@Psidom，你能帮我解决我的条件计数问题吗？ *.com/questions/64470031/…

【解决方案2】：

根据@Psidom 的回答，我的回答如下

from pyspark.sql.functions import col,when,count

test.groupBy("x").agg(
    count(when(col("y") > 12453, True)),
    count(when(col("z") > 230, True))
).show()

【讨论】：

请注意，此处的 True 值不是必需的 - 任何非 null 值都会达到相同的结果，因为 count() 计数非 null。

【解决方案3】：

count 函数跳过 null 值，所以你可以试试这个：

import pyspark.sql.functions as F

def count_with_condition(cond):
    return F.count(F.when(cond, True))

【讨论】：