PySpark 中对数组列计算均值与众数并新增字段的完整教程

本文介绍如何在 Pyspark DataFrame 中高效、稳定地为 Array 类型的数值列计算均值、为 array 类型的分类列计算众数，并将结果作为新列添加，避免 UDF 常见的序列化与类型错误。

本文介绍如何在 pyspark dataframe 中高效、稳定地为 array 类型的数值列计算均值、为 array 类型的分类列计算众数，并将结果作为新列添加，避免 udf 常见的序列化与类型错误。

在 PySpark 中直接对数组列（如 score: array 或 review: array）进行聚合统计（如均值、众数）时，若盲目使用 Python 原生 statistics 模块配合 UDF，极易因类型不匹配、空数组、序列化失败或 jvm/Python 环境隔离等问题导致运行时报错（如 PicklingError、NullPointerException 或 Calculation failed）。根本原因在于：原始 schema 中 score 列被错误定义为 array（含字符串元素），而实际数据是数字字符串（如 “83.52”），需显式转换；同时 statistics.mode() 在存在多个众数时会抛出 StatisticsError，且 UDF 缺乏容错机制。

✅ 推荐方案：组合内置函数 + 安全 UDF，兼顾性能与健壮性

1. 正确声明 Schema，确保类型安全

首先应明确定义 score 为 ArrayType(doubleType())，而非 String —— 这能避免后续手动 eval() 或 Float() 转换引发的 UDF 不稳定问题：

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, ArrayType, DoubleType, StringType from pyspark.sql.functions import col, explode, avg, udf from pyspark.sql import DataFrame  spark = SparkSession.builder.appName("ArrayAgg").getOrCreate()  # ✅ 正确定义 schema：score 是 double 数组，非 string 数组 schema = StructType([     StructField("id", IntegerType(), False),     StructField("score", ArrayType(DoubleType(), True), True),     StructField("review", ArrayType(StringType(), True), True) ])  # 示例数据（注意：score 元素为 float，非字符串） data = [     (1, [83.52, 81.79, 84.0, 75.0], ["P", "N", "P", "P"]),     (2, [86.13, 85.48], ["N", "N", "N", "P"]) ] df = spark.createDataFrame(data, schema)

2. 使用 explode + groupBy + agg 计算数组均值（零 UDF，高性能）

对数值数组求均值，优先使用内置 SQL 函数链，避免 UDF 开销与风险：

from pyspark.sql.functions import explode, avg, col, row_number from pyspark.sql.window import Window  # 为每行生成唯一标识（防止 groupBy 合并不同 id） df_with_id = df.withColumn("row_id", monotonically_increasing_id())  # 展开 score → 按 row_id 分组求均值 → 保留原结构 df_mean = (df_with_id            .withColumn("score_exploded", explode("score"))            .groupBy("row_id", "id", "score", "review")            .agg(avg("score_exploded").alias("scoreMean"))            .drop("row_id"))  # 清理临时列

3. 编写健壮的众数 UDF（处理空数组、多众数等边界）

statistics.mode() 不适用于多众数场景（如 [“P”,”N”,”P”,”N”]），且对空数组抛异常。以下 UDF 显式处理：

from collections import Counter from pyspark.sql.types import StringType  def safe_mode(arr):     if not arr or len(arr) == 0:         return None     # 统计频次，取最高频次的首个元素（稳定返回）     counter = Counter(arr)     max_count = max(counter.values())     # 返回第一个达到最高频次的元素（保证确定性）     for item in arr:         if counter[item] == max_count:             return item     return None  # 理论上不会到达  mode_udf = udf(safe_mode, StringType()) df_result = df_mean.withColumn("reviewMode", mode_udf(col("review")))

4. 最终结果与验证

df_result.select(     "id", "score", "review",      col("scoreMean").cast("decimal(10,2)").alias("scoreMean"),      "reviewMode" ).show(truncate=False)

输出：

+---+---------------------+----------+---------+----------+ |id |score                |review    |scoreMean|reviewMode| +---+---------------------+----------+---------+----------+ |1  |[83.52, 81.79, 84.0, 75.0]|[P, N, P, P]|81.08    |P         | |2  |[86.13, 85.48]       |[N, N, N, P]|85.81    |N         | +---+---------------------+----------+---------+----------+

⚠️ 关键注意事项

• 永远显式定义数组元素类型：ArrayType(DoubleType()) 比 ArrayType(StringType()) 更安全，避免 UDF 内部类型转换。
• 慎用 statistics.mode()：PySpark 3.4+ 推荐改用 Counter 手动实现，确保空输入、多众数、None 值兼容。
• 避免在 UDF 中调用 eval() 或 json.loads()：它们破坏序列化安全性，且无法跨 JVM/Python 进程传递。
• 性能提示：explode + groupBy 对大数据量可能产生倾斜，若数组极长，可考虑 aggregate 高阶函数（Spark 3.4+）替代：

  from pyspark.sql.functions import aggregate, col df.withColumn("scoreMean",      aggregate("score",          struct(lit(0.0).alias("sum"), lit(0).alias("count")),          lambda acc, x: struct(acc["sum"] + x, acc["count"] + 1),         lambda acc: acc["sum"] / acc["count"]     ) )

通过类型预校验、内置函数优先、UDF 边界防护三重策略，即可稳定、高效地完成数组列的均值与众数计算任务。