Pandas转PySpark时DataFrame无percentile属性报错如何解决

报错核心原因是PySpark DataFrame没有内置percentile/quantile实例方法，分位数计算属于聚合类操作，不能像Pandas那样直接在列或DataFrame对象上调用，必须通过内置聚合函数、或者DataFrame提供的分位数统计接口传入执行。
原代码除了调用方法不存在的问题，还有两个逻辑缺陷：

• 逐列循环写聚合逻辑会重复扫描全表，Spark懒执行机制下会生成多个冗余执行计划，数据量大时性能极差
• 循环中直接append列计算表达式拿到的是未执行的Column对象，不是实际计算值，就算方法存在最后拼接pandas DataFrame时也会出现类型错误

以下代码完全对齐原Pandas逻辑，且做了性能优化，一次性完成所有目标列的分位数计算：

from pyspark.sql import functions as F
import pandas as pd

# 1. 获取需要计算分位数的目标列列表，过滤不存在的列避免运行报错
target_cols = spark_df.select('dic').rdd.flatMap(lambda x: x).collect()
valid_cols = [col for col in target_cols if col in df.columns]

# 2. 一次性聚合计算所有目标列的0.9分位数
# 大表场景用percentile_approx性能更好，需要精确结果可替换为percentile
agg_expressions = [F.percentile_approx(col, 0.9).alias(col) for col in valid_cols]
quantile_result = df.agg(*agg_expressions).collect()[0]

# 3. 拼接为和原Pandas输出结构一致的结果
df_1 = pd.DataFrame({
    'dic': valid_cols,
    'Percentile': [quantile_result[col] for col in valid_cols]
})

如果需要返回Spark DataFrame而非Pandas DataFrame，把最后一步替换为以下代码即可：

df_1 = spark.createDataFrame(
    zip(valid_cols, [quantile_result[col] for col in valid_cols]),
    schema=['dic', 'Percentile']
)

也可以直接用PySpark DataFrame内置的approxQuantile接口实现，第三个参数为允许的相对误差，设为0时为精确计算：

# 相对误差设为0.001时计算速度会大幅提升，误差可满足绝大多数业务场景
quantile_list = df.approxQuantile(valid_cols, [0.9], 0.0)
df_1 = pd.DataFrame({
    'dic': valid_cols,
    'Percentile': [val[0] for val in quantile_list]
})

• 不要直接照搬Pandas的方法调用习惯，PySpark中所有跨行的统计计算（分位数、求和、均值等）都需要通过聚合操作、或者专用统计接口实现，不存在Pandas Series/DF上挂载的同名方法
• 尽量避免逐行、逐列循环触发Spark计算，能批量聚合的逻辑尽量合并为单次作业，减少全表扫描次数

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