PySpark

Question

目前我正在使用 UDF 收集前 5 个最常见的值。

目标是获得相同的结果without using UDF并拥有最有效的解决方案（避免循环中的 groupBy）。

这是我用来获得结果的代码：

from pyspark.sql import functions as F

df = df.select('A', 'B', ...)

@F.udf
def get_top_5_udf(x)
    from collections import Counter
    return [elem[0] for elem in Counter(x).most_common(5)]

agg_expr = [get_top_5_udf(F.collect_list(col)).alias(col) for col in df.columns]

df_top5 = df.agg(*agg_expr)

结果如下所示：

# result
#+-----------------+--------------+---------------+
#|         A       |       B      |      ...      |
#+-----------------+--------------+---------------+
#| [1, 2, 3, 4, 5] |     [...]    |      ...      |
#+-----------------+--------------+---------------+

Answer 1

您可以在聚合之前尝试使用按每列分区的 window 计数：

from pyspark.sql import functions as F, Window

result = df.select(*[
    F.struct(
        F.count(c).over(Window.partitionBy(c)).alias("cnt"),
        F.col(c).alias("val")
    ).alias(c) for c in df.columns
]).agg(*[
    F.slice(
        F.expr(f"transform(sort_array(collect_set({c}), false), x -> x.val)"),
        1, 5
    ).alias(c) for c in df.columns
])

result.show()

PySpark - 获取每列的前 5 个最常见值（无 UDF）

PySpark - Get top 5 most frequent values for every column (without UDF)

dataframe

apache-spark

apache-spark-sql