如何在Databricks中对数据集做winsorize处理并修复index-1越界错误

你遇到的报错本质是调用scipy.stats.mstats.winsorize时传入的序列长度为0，winsorize内部计算上下分位点时需要访问序列首尾元素，空序列就会触发下标越界错误。
触发空序列的具体代码问题：

• 内层循环中你重复从全局df_read按site、utility、tag三个维度过滤，部分组合可能没有对应数据，得到空的df_r
• outlier_v2函数直接使用全局作用域的df_r，没有做非空校验就直接调用winsorize
• 额外潜在bug：你没有将处理后的df_r存入结果列表，最后使用的results变量未定义，会二次报错

1. 修改outlier_v2为显式接收DataFrame参数，增加空值判断逻辑
2. 每次过滤后先校验df_r非空再做后续处理
3. 处理完成后将结果追加到结果列表，最后合并为完整数据集供Spark调用

修复后完整代码 #

from pandasql import sqldf
import pandas as pd
from scipy.stats.mstats import winsorize

# 提取site和utility唯一组合
q="""SELECT DISTINCT site, utility FROM df_read;"""
pysqldf = lambda q: sqldf(q, globals())
a_df = pysqldf(q)
a_ls = a_df.values.tolist()

# 绘图函数保留，你按需调用即可
def plot_df(df, x, y,y1,y2,y3, title="", xlabel='Date', ylabel='Value', dpi=100):
    import matplotlib.pyplot as plt
    plt.figure(figsize=(16,5), dpi=dpi)
    plt.plot(x, y, color='tab:blue')
    plt.plot(x,y1,color='tab:red')
    plt.plot(x,y2,color='tab:red')
    plt.plot(x,y3,color='tab:red')
    plt.gca().set(title=title, xlabel=xlabel, ylabel=ylabel)
    plt.show()

# 改造后的异常值处理函数
def outlier_v2(df): 
    # 空df直接返回
    if df.empty:
        return df
    df = df.reset_index(drop=True)
    df["converted_value_adjusted"] = winsorize(df['converted_value'], limits=[0.00, 0.05])
    return df

# 循环处理所有分组
df_all = []
for entry in a_ls:
    site, utility = entry[0], entry[1]
    # 先取当前site+utility下的所有tag
    temp_df = df_read[(df_read["site"]==site) & (df_read["utility"] == utility)].copy()
    if temp_df.empty:
        continue
    tags = temp_df['tag'].unique()
    for tag_v in tags:
        # 从已经过滤过的temp_df里取对应tag数据，减少全局过滤的空数据概率
        df_r = temp_df[temp_df["tag"] == tag_v].copy()
        # 处理后追加到结果列表
        processed_df = outlier_v2(df_r)
        df_all.append(processed_df)

# 合并所有处理结果
results = pd.concat(df_all, ignore_index=True)
# 转Spark DataFrame
spark_results = spark.createDataFrame(results)
display(spark_results)

可选优化建议 #

• 如果数据量较大，不需要用pandasql取唯一组合，可以直接用pandas自带的df_read[['site','utility']].drop_duplicates().values.tolist()实现，性能更好
• 如果要保留更多原数据的索引，可自行修改reset_index的参数逻辑

内容的提问来源于stack exchange，提问作者n00b data engineer