PySpark 计算PySpark DataFrame列的众数

在本文中，我们将介绍如何使用PySpark计算一个PySpark DataFrame列的众数。众数是指数据集中出现次数最多的值。通过计算众数，我们可以了解到数据集中的最常见取值。

方法一：使用groupBy和agg函数

最简单的方法是使用groupBy和agg函数来计算众数。groupBy函数用于按照指定的列对数据进行分组，而agg函数用于对分组后的数据进行聚合操作。

# 导入相关库
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 创建SparkSession
spark = SparkSession.builder.getOrCreate()

# 创建示例数据
data = [("Alice", 25), ("Bob", 35), ("Alice", 30), ("Bob", 35), ("Alice", 30)]
schema = StructType([StructField("Name", StringType(), True), StructField("Age", IntegerType(), True)])
df = spark.createDataFrame(data, schema)

# 计算众数
mode_df = df.groupBy("Age").agg(count("*").alias("count")).orderBy(desc("count")).limit(1)
mode = mode_df.select("Age").collect()[0][0]

# 打印众数
print("Mode of Age column is:", mode)

在上述代码中，我们创建了一个包含姓名和年龄的示例数据集。然后，我们使用groupBy和agg函数将数据按照年龄进行分组，并计算每个年龄分组的计数。最后，我们对计数进行降序排序并限制结果为1行，从而得到众数。最后，我们通过mode_df.select(“Age”).collect()[0][0]获取众数值，并打印输出。

方法二：使用pandas_udf函数

如果需要对Spark DataFrame进行更复杂的操作，可以使用pandas_udf函数来自定义函数，并利用pandas库的功能计算众数。pandas_udf函数能够将Spark DataFrame转换为pandas DataFrame，然后使用pandas库对数据进行处理。

# 导入相关库
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 导入pandas
import pandas as pd

# 创建SparkSession
spark = SparkSession.builder.getOrCreate()

# 创建示例数据
data = [("Alice", 25), ("Bob", 35), ("Alice", 30), ("Bob", 35), ("Alice", 30)]
schema = StructType([StructField("Name", StringType(), True), StructField("Age", IntegerType(), True)])
df = spark.createDataFrame(data, schema)

# 自定义函数计算众数
@pandas_udf(IntegerType())
def mode_udf(age: pd.Series) -> int:
    return int(age.mode())

# 计算众数
mode = df.agg(collect_list("Age").alias("Age")).selectExpr("mode_udf(Age) as mode").collect()[0]["mode"]

# 打印众数
print("Mode of Age column is:", mode)

在上述代码中，我们首先导入pandas库以支持使用pandas_udf函数。然后，我们创建了一个自定义函数mode_udf，用于计算pandas DataFrame的众数，并将返回值的数据类型设置为整型。接下来，我们使用agg函数和collect_list函数将Spark DataFrame中的Age列以List的形式收集起来，并将其命名为Age。然后，我们使用selectExpr函数调用mode_udf函数，将Age作为参数传递给mode_udf函数，并将结果命名为mode。最后，我们通过selectExpr("mode_udf(Age) as mode").collect()[0]["mode"]获取众数值，并打印输出。