如何在PySpark中提取DataFrame列表列内出现频率最高的元素

以下提供两种兼容不同Spark版本的实现方案，均可直接落地：

方案1：explode + 窗口排序（兼容所有Spark版本，逻辑直观） #

无需高阶函数支持，逻辑易懂，适合新手使用：
首先导入依赖函数：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

模拟样例数据（实际使用时可替换为自己的DataFrame）：

data = [
    ("X", ["Element5", "Element1", "Element5"]),
    ("Y", ["Element Unknown", "Element Unknown", "Element_Z"])
]
df = spark.createDataFrame(data, schema=["ID", "Elements"])

核心处理逻辑：

res_df = df.withColumn("single_elem", F.explode("Elements")) \
    # 按ID、原数组、单个元素分组，统计每个元素出现次数
    .groupBy("ID", "Elements", "single_elem").agg(F.count("*").alias("freq")) \
    # 按ID分组，按元素频率倒序排序，取排序第一的元素
    .withColumn("rank", F.row_number().over(Window.partitionBy("ID").orderBy(F.desc("freq")))) \
    .filter(F.col("rank") == 1) \
    .select("ID", "Elements", F.col("single_elem").alias("Output_column"))

注意：如果存在多个元素频率相同的情况，row_number()会随机取第一个，如需返回所有最高频元素，可将row_number()替换为rank()，后续再用collect_list合并为数组即可。

方案2：高阶函数实现（Spark 2.4+支持，无Shuffle性能更高） #

无需打散数据，行内直接计算，性能远高于方案1，适合大数据量场景：

res_df = df.withColumn("Output_column",
    F.expr("""
        aggregate(
            Elements,
            cast(map() as map<string, int>),
            -- 遍历数组统计每个元素的出现频率，生成频率映射表
            (acc, current_elem) -> map_concat(acc, map(current_elem, coalesce(acc[current_elem], 0) + 1)),
            -- 提取频率最高的元素
            freq_map -> map_keys(freq_map)[array_position(map_values(freq_map), array_max(map_values(freq_map))) - 1]
        )
    """)
)

内容的提问来源于stack exchange，提问作者Droid-Bird