Spark DataFrame：计算每列的不同值答案

【问题标题】：Spark DataFrame: count distinct values of every columnSpark DataFrame：计算每列的不同值
【发布时间】：2017-04-14 19:27:12
【问题描述】：

问题几乎就在标题中：是否有一种有效的方法来计算 DataFrame 中每一列中的不同值？

describe 方法仅提供计数但不提供非重复计数，我想知道是否有一种方法可以获取所有（或某些选定的）列的非重复计数。

【问题讨论】：

标签： apache-spark apache-spark-sql distinct-values

【解决方案1】：

在pySpark 中，您可以使用countDistinct() 执行类似的操作：

from pyspark.sql.functions import col, countDistinct

df.agg(*(countDistinct(col(c)).alias(c) for c in df.columns))

Scala 类似：

import org.apache.spark.sql.functions.countDistinct
import org.apache.spark.sql.functions.col

df.select(df.columns.map(c => countDistinct(col(c)).alias(c)): _*)

如果您想在可能降低准确性的情况下加快速度，您也可以使用approxCountDistinct()。

【讨论】：

您能解释一下这里的* 是什么（在您的 pyspark 解决方案中）吗？
Python 中的星号运算符可用于从迭代器中解包参数以进行函数调用，另见here。

【解决方案2】：

计算多个聚合将非常昂贵。我建议您改用近似方法。在这种情况下，近似不同的计数：

val df = Seq((1,3,4),(1,2,3),(2,3,4),(2,3,5)).toDF("col1","col2","col3")

val exprs = df.columns.map((_ -> "approx_count_distinct")).toMap
df.agg(exprs).show()
// +---------------------------+---------------------------+---------------------------+
// |approx_count_distinct(col1)|approx_count_distinct(col2)|approx_count_distinct(col3)|
// +---------------------------+---------------------------+---------------------------+
// |                          2|                          2|                          3|
// +---------------------------+---------------------------+---------------------------+

approx_count_distinct 方法在底层依赖于 HyperLogLog。

HyperLogLog 算法及其变体 HyperLogLog++（在 Spark 中实现）依赖于以下聪明观察。

如果数字在一个范围内均匀分布，则不同元素的计数可以从数字的二进制表示中前导零的最大数量来近似。

例如，如果我们观察到一个数字，其二进制形式的数字是0…(k times)…01…1，那么我们可以估计集合中有大约 2^k 个元素。这是一个非常粗略的估计，但可以使用草图算法将其精炼到非常精确的程度。

可以在original paper 中找到对该算法背后机制的全面解释。

注意：从 Spark 1.6 开始，当 Spark 调用 SELECT SOME_AGG(DISTINCT foo)), SOME_AGG(DISTINCT bar)) FROM df 时，每个子句都应该为每个子句触发单独的聚合。而这与我们聚合一次的SELECT SOME_AGG(foo), SOME_AGG(bar) FROM df 不同。因此，使用count(distinct(_)) 和approxCountDistinct（或approx_count_distinct）时，性能将无法比拟。

这是自 Spark 1.6 以来的行为变化之一：

使用针对具有不同聚合的查询的改进查询计划器 (SPARK-9241)，具有单个不同聚合的查询计划已更改为更强大的版本。要切换回 Spark 1.5 的 planner 生成的计划，请将 spark.sql.specializeSingleDistinctAggPlanning 设置为 true。 (SPARK-12077)

参考：Approximate Algorithms in Apache Spark: HyperLogLog and Quantiles.

【讨论】：

请注意：请注意，对于几乎每个值都是唯一的列，在默认配置中 approx_count_distinct 可能会产生高达 10% 的错误，并且实际上可能与 count_distinct 花费相同的时间。它甚至可能返回一个高于实际行数的值。
没错，但你的数据集越大，误差越小。

【解决方案3】：

如果您只想计算特定列，那么以下可能会有所帮助。虽然它的答案迟了。它可能会帮助某人。（pyspark 2.2.0 已测试）

from pyspark.sql.functions import col, countDistinct
df.agg(countDistinct(col("colName")).alias("count")).show()

【讨论】：

如果您希望将答案放在变量中，而不是向用户显示，请将 .show() 替换为 .collect()[0][0]

【解决方案4】：

可以使用SQL的count(column name)函数

或者，如果您正在使用数据分析并且想要粗略估计而不是每列的精确计数，您可以使用 approx_count_distinct 函数 approx_count_distinct(expr[, relativeSD])

【讨论】：

【解决方案5】：

添加到 desaiankitb 的答案，这将为您提供更直观的答案：

from pyspark.sql.functions import count

df.groupBy(colname).count().show()

【讨论】：

对于那些希望显示所选列中出现的每个唯一值的计数的人来说，这是一个很好的答案。