PySpark与Pandas中基于列均值筛选并排序DataFrame列的实现

给定如下数据与初始DataFrame：

data = [[12, 112, 14],
        [120, 112, 114],
        [88, 92, 74],
        [17, 118, 133],
        [19, 19, 14],
        [11, 12, 14]]
columns = ['Subject_1', 'Subject_2', 'Subject_3']

需求：生成DF2，需满足以下条件：

1. 仅保留列均值大于指定阈值的列；
2. 若符合条件1的列数量过多（可自定义阈值），仅保留列均值处于75分位数及以上的列；
3. 最终列按均值升序排列。

PySpark 实现方案 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import mean
import numpy as np

# 初始化Spark会话
spark = SparkSession.builder.appName("SubjectFilter").getOrCreate()
# 创建初始DataFrame
dataframe = spark.createDataFrame(data, columns)

# 自定义参数（可按需调整）
mean_threshold = 60  # 列均值达标阈值
max_allowed_cols = 2  # 定义"数量过多"的临界值

# 1. 计算所有列的均值
col_means_df = dataframe.agg(*[mean(col).alias(col) for col in dataframe.columns])
col_means = col_means_df.collect()[0].asDict()

# 2. 筛选均值大于阈值的列
filtered_cols = [col for col in col_means if col_means[col] > mean_threshold]

# 3. 若列数超过阈值，进一步筛选75分位数以上的列
if len(filtered_cols) > max_allowed_cols:
    mean_values = [col_means[col] for col in filtered_cols]
    q75 = np.percentile(mean_values, 75)
    filtered_cols = [col for col in filtered_cols if col_means[col] >= q75]

# 4. 按均值升序排列列并生成DF2
sorted_cols = sorted(filtered_cols, key=lambda x: col_means[x])
df2 = dataframe.select(*sorted_cols)
df2.show()

说明 #

• 通过agg+mean批量计算列均值，转字典便于后续筛选操作；
• 用numpy.percentile计算75分位数，需确保环境已安装numpy；
• 所有参数可根据实际业务需求灵活调整。

Pandas 实现方案 #

import pandas as pd
import numpy as np

# 创建初始Pandas DataFrame
df = pd.DataFrame(data, columns=columns)

# 自定义参数（可按需调整）
mean_threshold = 60
max_allowed_cols = 2

# 1. 计算各列均值
col_means = df.mean().to_dict()

# 2. 筛选均值大于阈值的列
filtered_cols = [col for col in col_means if col_means[col] > mean_threshold]

# 3. 列数过多时保留75分位数以上的列
if len(filtered_cols) > max_allowed_cols:
    mean_values = [col_means[col] for col in filtered_cols]
    q75 = np.percentile(mean_values, 75)
    filtered_cols = [col for col in filtered_cols if col_means[col] >= q75]

# 4. 按均值升序排列列并生成DF2
sorted_cols = sorted(filtered_cols, key=lambda x: col_means[x])
df2 = df[sorted_cols]
print(df2)

说明 #

• Pandas的mean()方法直接返回列均值，操作更简洁；
• 逻辑与PySpark版本完全对齐，参数可统一调整；
• 最终输出可直接打印或按需保存。

内容的提问来源于stack exchange，提问作者user2458922