使用pyspark计算groupBy的总数百分比

Question

我在pyspark中有以下代码，它会生成一个表格，显示某一列的不同值及其计数。我想要另外一列显示每行所代表的总计数的百分比是多少。我该如何实现？

difrgns = (df1
           .groupBy("column_name")
           .count()
           .sort(desc("count"))
           .show())

提前致谢！

- Giordan Pretelin

---

这里有一个最近的自问自答问题，对你可能有帮助这里。 - pault

Answer 1

如果不喜欢使用窗口方式，可以考虑以下替代示例。评论中提到这是更好的方法：

# Running in Databricks, not all stuff required
from pyspark.sql import Row
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import *
#from pyspark.sql.functions import col

data = [("A", "X", 2, 100), ("A", "X", 7, 100), ("B", "X", 10, 100),
        ("C", "X", 1, 100), ("D", "X", 50, 100), ("E", "X", 30, 100)]
rdd = sc.parallelize(data)

someschema = rdd.map(lambda x: Row(c1=x[0], c2=x[1], val1=int(x[2]), val2=int(x[3])))

df = sqlContext.createDataFrame(someschema)

tot = df.count()

df.groupBy("c1") \
  .count() \
  .withColumnRenamed('count', 'cnt_per_group') \
  .withColumn('perc_of_count_total', (F.col('cnt_per_group') / tot) * 100 ) \
  .show()

返回：

 +---+-------------+-------------------+
| c1|cnt_per_group|perc_of_count_total|
+---+-------------+-------------------+
|  E|            1| 16.666666666666664|
|  B|            1| 16.666666666666664|
|  D|            1| 16.666666666666664|
|  C|            1| 16.666666666666664|
|  A|            2|  33.33333333333333|
+---+-------------+-------------------+

我专注于Scala，使用它似乎更容易。话虽如此，评论中建议的解决方案使用了Window，在Scala中我也会使用over()函数。

- thebluephantom

---

我认为你需要先广播你的 tot 变量。 - Sergey Zakharov

根据我从实践中所看到的，这似乎是可行但并非强制性的。如果有错误会被注意到。 - thebluephantom

Answer 2

当 df 本身是一个更复杂的转换链，运行两次 —— 首先计算总数，然后分组并计算百分比 —— 太昂贵时，可以利用窗口函数来实现类似的结果。这里有一个更广义的代码（扩展了bluephantom的答案），可用于许多按组维度进行分组：

from pyspark.sql import Row
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.window import Window

data = [("A", "X", 2, 100), ("A", "X", 7, 100), ("B", "X", 10, 100),
        ("C", "X", 1, 100), ("D", "X", 50, 100), ("E", "X", 30, 100)]
rdd = sc.parallelize(data)

someschema = rdd.map(lambda x: Row(c1=x[0], c2=x[1], val1=int(x[2]), val2=int(x[3])))

df = (sqlContext.createDataFrame(someschema)
      .withColumn('total_count', count('*').over(Window.partitionBy(<your N-1 dimensions here>)))
     .groupBy(<your N dimensions here>)
       .agg((count('*')/first(col('total_count'))).alias('percent_total'))
)

df.show()

- Dmitry B.

---

使用窗口函数是一个有趣的想法，但你确定你的示例代码正确吗？我尝试将 <your dimensions here> 设置为 "c1" 并运行它，每个 c1 的值都返回了 1.0 的 percent_total。 - Nick Chammas

@NickChammas：是的，模板代码是正确的，但是你必须小心尺寸：总计数是在不同级别上完成的，而不是分组，即如果使用N个维度进行分组，则窗口函数必须按N-1个维度进行分区。对于OP的特定示例，如果groupBy（'c1'），则使用没有参数的Window.partitionBy（）。 - Dmitry B.

你如何在普通的Spark SQL中使用无参数的partitionBy函数？ - Dalupus

@Dalupus：尝试构建一个PySpark查询，然后在其上运行“explain”。如果我没记错的话，这应该会给出相应的SQL语句。 - Dmitry B.

Answer 3

您可以使用 groupby 和 agg 进行分组和聚合。例如，对于以下 DataFrame：

+--------+-----+
|category|value|
+--------+-----+
|       a|    1|
|       b|    2|
|       a|    3|
+--------+-----+

您可以使用：

import pyspark.sql.functions as F

df.groupby('category').agg(
    (F.count('value')).alias('count'),
    (F.count('value') / df.count()).alias('percentage')
).show()

输出：

+--------+-----+------------------+
|category|count|        percentage|
+--------+-----+------------------+
|       b|    1|0.3333333333333333|
|       a|    2|0.6666666666666666|
+--------+-----+------------------+

或者，您可以使用SQL：

df.createOrReplaceTempView('df')

spark.sql(
    """
    SELECT category,
           COUNT(*) AS count,
           COUNT(*) / (SELECT COUNT(*) FROM df) AS ratio
    FROM df
    GROUP BY category
    """
).show()

- Mykola Zotko

Answer 4

更加美观的输出结果，消除多余小数并进行排序。

import pyspark.sql.functions as func
count_cl = data_fr.count()

data_fr \
.groupBy('col_name') \
.count() \
.withColumn('%', func.round((func.col('count')/count_cl)*100,2)) \
.orderBy('count', ascending=False) \
.show(4, False)

    +--------------+-----+----+
    | col_name     |count|   %|
    +--------------------+----+
    |      C.LQQQQ |30957|8.91|
    |      C.LQQQQ |29688|8.54|
    |      C-LQQQQ |29625|8.52|
    |       CLQQQQ |29342|8.44|    
    +--------------------+----+

- s510

---

这里的 count_c1 是什么？ - Evan