PySpark 中的 RobustScaler

Posted 2023-04-15

【中文标题】PySpark 中的 RobustScaler

【英文标题】：RobustScaler in PySpark

【发布时间】：2020-01-24 10:45:42

【问题描述】：

我想使用 RobustScaler 来预处理数据。在 sklearn 中可以找到

sklearn.preprocessing.RobustScaler

。但是，我使用的是 pyspark，所以我尝试使用以下命令导入它：

 from pyspark.ml.feature import RobustScaler

但是，我收到以下错误：

ImportError: cannot import name 'RobustScaler' from 'pyspark.ml.feature' 

正如 pault 指出的，RobustScaler 仅在 pyspark 3 中实现。我正在尝试将其实现为：

class PySpark_RobustScaler(Pipeline):
    def __init__(self):
        pass

    def fit(self, df):
        return self

    def transform(self, df):
        self._df = df
        for col_name in self._df.columns:
            q1, q2, q3 = self._df.approxQuantile(col_name, [0.25, 0.5, 0.75], 0.00)
            self._df = self._df.withColumn(col_name, 2.0*(sf.col(col_name)-q2)/(q3-q1))
        return self._df

arr = np.array(
            [[ 1., -2.,  2.],
            [ -2.,  1.,  3.],
            [ 4.,  1., -2.]]
          )

rdd1 = sc.parallelize(arr)
rdd2 = rdd1.map(lambda x: [int(i) for i in x])
df_sprk = rdd2.toDF(["A", "B", "C"])
df_pd = pd.DataFrame(arr, columns=list('ABC'))

PySpark_RobustScaler().fit(df_sprk).transform(df_sprk).show()
print(RobustScaler().fit(df_pd).transform(df_pd))

但是我发现要获得与 sklearn 相同的结果，我必须将结果乘以 2。此外，我担心如果一列有许多接近零的值，则四分位间距 q3-q1 可能会变得太小并让结果发散，创建空值。

有人对如何改进它有任何建议吗？

【问题讨论】：

好像这在version 3 中可用，请参阅SPARK-28399。

【参考方案1】：

此功能已在recent pyspark 版本中发布。