PySpark与Sklearn预处理对照：从Sklearn迁移到PySpark的避坑指南

1. 为什么要在PySpark和Sklearn之间做预处理对照做机器学习项目的人迟早会撞上一个尴尬的现实本地用Sklearn跑得好好的预处理流水线一旦数据量涨到单机内存扛不住就得整体搬到PySpark上重写一遍。麻烦的地方不在于“会不会用PySpark”而在于两套API对同一件事的命名、默认行为和输出形态经常对不上。比如标准化Sklearn里叫StandardScalerPySpark里也有StandardScaler但前者默认按列算均值和标准差后者默认按列算均值和标准差的同时还要求输入是Vector类型再比如缺失值填充Sklearn的SimpleImputer默认用均值PySpark的Imputer默认也是均值但PySpark只支持数值列字符串列的缺失值处理得另想办法。我最初做这类迁移时习惯性地把Sklearn代码逐行翻译成PySpark结果踩了一堆坑StringIndexer和OneHotEncoder的配合顺序搞反、VectorAssembler把标签列也拼进去、StandardScaler的输出列名和原始列名冲突。后来我干脆把两套框架的预处理操作做成一张对照表每做一个操作就查一下“Sklearn里怎么做、PySpark里怎么做、差异在哪、什么情况下会出错”。这篇文章就是这张对照表的完整展开适合两类人看一类是刚接触PySpark、想把Sklearn经验迁移过去的开发者另一类是已经在用PySpark但经常被Vector类型和Pipeline阶段顺序搞晕的工程师。我会按预处理的实际操作顺序来讲每个操作都给出两边的代码、参数含义、默认行为差异以及我在真实项目里踩过的坑。提示本文所有代码示例都基于PySpark 3.x和Sklearn 1.x的稳定API不涉及任何需要额外安装的第三方扩展包。如果你用的是更早的版本部分参数名可能有差异建议先确认版本。2. 数据读取与类型推断两套框架的第一道分水岭2.1 Sklearn的“先读进来再说”与PySpark的“先定Schema再读”Sklearn本身没有数据读取能力它依赖pandas、numpy或者Python原生列表。你通常是这样开始的import pandas as pd df pd.read_csv(data.csv)pandas会自动推断每列的类型数值列变成int64或float64字符串列变成object。这个“自动推断”在数据量小的时候很方便但到了PySpark里如果你也用spark.read.csv(data.csv, inferSchemaTrue)Spark会额外扫一遍数据来推断类型对于几十GB的文件这一遍扫描可能就要好几分钟。更稳妥的做法是显式定义Schemafrom pyspark.sql.types import StructType, StructField, DoubleType, StringType schema StructType([ StructField(age, DoubleType(), True), StructField(income, DoubleType(), True), StructField(city, StringType(), True), StructField(label, DoubleType(), True) ]) df spark.read.csv(data.csv, headerTrue, schemaschema)显式Schema的好处不只是快更重要的是避免类型推断出错。我遇到过一列“收入”里混了几个空字符串pandas会把它整列变成objectSklearn的SimpleImputer直接报错而Spark的inferSchema可能把这一列推断成string后续VectorAssembler要求数值列时又报错。显式声明DoubleType之后空字符串会被转成nullImputer就能正常处理了。2.2 缺失值在两边长什么样这是最容易被忽略的差异。在pandas里缺失值通常是NaN浮点或None对象Sklearn的SimpleImputer能识别np.nan。在PySpark里缺失值统一表现为null但null在数值列和字符串列里的行为不一样数值列的null参与VectorAssembler时会被当成0吗不会VectorAssembler遇到null会直接抛异常除非你设置handleInvalidskip或handleInvalidkeep。字符串列的null在StringIndexer里默认会被分配一个额外的索引通常是numLabels这个行为在Sklearn的LabelEncoder里是不存在的LabelEncoder遇到NaN会直接报错。我一般会在读取之后立刻做一次缺失值统计两边的方法也不一样# Sklearn/pandas df.isnull().sum() # PySpark from pyspark.sql.functions import col, count, when df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()PySpark这个写法看起来啰嗦但它是列式扫描比逐列filter再count快得多。统计完之后你才能决定每一列是用均值填充、中位数填充还是直接删除。3. 缺失值填充SimpleImputer与Imputer的默认行为差异3.1 数值列填充的对照Sklearn的SimpleImputer用法很直接from sklearn.impute import SimpleImputer imputer SimpleImputer(strategymean) df[[age, income]] imputer.fit_transform(df[[age, income]])PySpark的Imputer需要指定输入列、输出列和策略from pyspark.ml.feature import Imputer imputer Imputer( inputCols[age, income], outputCols[age_imputed, income_imputed], strategymean ) model imputer.fit(df) df model.transform(df)差异点有三个。第一PySpark的Imputer不会覆盖原始列它生成新列这意味着你后续要么把原始列删掉要么在VectorAssembler里用新列。第二PySpark的Imputer只支持数值列如果你把字符串列传进去它会直接报IllegalArgumentException。第三PySpark的Imputer在fit阶段计算均值时会忽略null但如果你在transform之后又引入了新的null比如从其他表join进来它不会重新计算因为fit已经结束了。注意PySpark的Imputer在策略为mean或median时如果某一列全是null它会抛异常。Sklearn的SimpleImputer在这种情况下会抛ValueError并提示“不能对全空列计算均值”。两边都会报错但错误信息不同排查时别被误导。3.2 字符串列缺失值的处理思路Sklearn里可以用SimpleImputer(strategymost_frequent)处理字符串列PySpark的Imputer不支持字符串。我的做法是先用fillna把字符串列的null替换成一个固定值比如unknown然后再走StringIndexerdf df.fillna({city: unknown})这个unknown会成为一个独立的类别在StringIndexer里被编码成一个索引。如果你不想让它成为一个类别那就得在fillna之后用filter把这一行删掉但删行会损失数据需要权衡。3.3 一个容易踩的坑填充后分布偏移用均值填充数值列会让这一列的方差变小因为所有缺失值都被拉到了均值附近。如果缺失比例很高比如超过30%填充后的分布和原始分布差异会很大模型学到的模式可能失真。我一般会先看缺失比例超过20%的列就考虑加一个“是否缺失”的指示列让模型自己判断这个缺失是否有信息量。Sklearn里可以用MissingIndicatorPySpark里得手动加from pyspark.sql.functions import when, lit df df.withColumn(income_missing, when(col(income).isNull(), 1).otherwise(0))这个指示列在后续VectorAssembler里和填充后的income一起拼进去模型就能同时看到“值是多少”和“这个值是不是填出来的”。4. 类别编码StringIndexerOneHotEncoder与LabelEncoderOneHotEncoder的配合逻辑4.1 Sklearn的编码流程Sklearn处理类别特征通常有两种方式。如果类别之间没有顺序关系用OneHotEncoderfrom sklearn.preprocessing import OneHotEncoder encoder OneHotEncoder(handle_unknownignore, sparseFalse) encoded encoder.fit_transform(df[[city]])如果类别有顺序关系比如“低、中、高”用OrdinalEncoder或者手动映射。LabelEncoder在Sklearn里是给标签用的不是给特征用的这一点很多人会搞混。4.2 PySpark的编码流程PySpark的流程是固定的两步先StringIndexer把字符串列转成数值索引再OneHotEncoder把索引转成稀疏向量。顺序不能反因为OneHotEncoder的输入必须是数值索引列。from pyspark.ml.feature import StringIndexer, OneHotEncoder indexer StringIndexer(inputColcity, outputColcity_index, handleInvalidkeep) encoder OneHotEncoder(inputColcity_index, outputColcity_vec) pipeline Pipeline(stages[indexer, encoder]) model pipeline.fit(df) df model.transform(df)StringIndexer的handleInvalid参数有三个取值error默认遇到新类别报错、skip跳过含新类别的行、keep把新类别放到一个额外的索引里。在训练集上fit之后如果测试集出现了训练集没见过的类别error会直接让任务失败skip会丢数据keep会保留但索引是训练集类别数之外的一个值。我一般用keep然后在OneHotEncoder里设置dropLastTrue来避免多重共线性。4.3 两边编码结果的形态差异Sklearn的OneHotEncoder返回的是密集矩阵如果sparseFalse或稀疏矩阵列数等于类别数。PySpark的OneHotEncoder返回的是Vector类型的稀疏向量列数等于类别数如果dropLastFalse或类别数减一如果dropLastTrue。这个Vector不能直接喂给Sklearn的模型反过来Sklearn的矩阵也不能直接喂给PySpark的模型。如果你在做两边的对照实验需要把Vector转成数组from pyspark.ml.functions import vector_to_array df df.withColumn(city_array, vector_to_array(city_vec))这个vector_to_array是PySpark 3.0之后才有的之前版本需要用UDF手动拆性能差很多。4.4 高频类别与低频类别的处理当某一列的类别数很多比如城市名有几百个OneHotEncoder会产生几百维的稀疏向量大部分维度都是0。Sklearn里可以用min_frequency参数把低频类别合并成一个“其他”类别PySpark的StringIndexer没有这个参数得手动做from pyspark.sql.functions import count city_counts df.groupBy(city).count().filter(col(count) 50) frequent_cities [row[city] for row in city_counts.collect()] df df.withColumn(city_grouped, when(col(city).isin(frequent_cities), col(city)).otherwise(lit(other)))这个阈值这里是50需要根据数据量和业务含义来定。我一般会先看类别分布的累积曲线找到覆盖80%样本的最小类别集合把剩下的归为“其他”。5. 数值标准化与归一化StandardScaler和MinMaxScaler的Vector依赖5.1 Sklearn的列式操作Sklearn的StandardScaler直接对DataFrame的数值列操作from sklearn.preprocessing import StandardScaler scaler StandardScaler() df[[age, income]] scaler.fit_transform(df[[age, income]])它返回的是numpy数组列的顺序和输入一致。fit阶段计算均值和标准差transform阶段做减均值除标准差。5.2 PySpark的Vector式操作PySpark的StandardScaler要求输入是一个Vector列通常由VectorAssembler生成from pyspark.ml.feature import VectorAssembler, StandardScaler assembler VectorAssembler(inputCols[age, income], outputColfeatures_raw) df assembler.transform(df) scaler StandardScaler(inputColfeatures_raw, outputColfeatures_scaled, withMeanTrue, withStdTrue) model scaler.fit(df) df model.transform(df)withMeanTrue表示减均值withStdTrue表示除标准差。如果两个都设为FalseStandardScaler什么都不做这个默认值容易让人误以为它会自动标准化。实际上PySpark的StandardScaler默认withMeanFalse、withStdTrue也就是只除标准差不减均值。Sklearn的StandardScaler默认两者都做。这个差异如果不注意两边跑出来的结果会对不上。5.3 归一化的对照Sklearn的MinMaxScaler默认把数据缩放到[0,1]区间。PySpark的MinMaxScaler也是缩放到[0,1]但同样要求Vector输入而且它的min和max参数默认是0和1你可以改成其他范围。from pyspark.ml.feature import MinMaxScaler scaler MinMaxScaler(inputColfeatures_raw, outputColfeatures_scaled, min0.0, max1.0)两边在遇到常数列所有值相同时的行为也不一样。Sklearn的MinMaxScaler会把常数列缩放到0因为分母是0它内部做了处理PySpark的MinMaxScaler在fit时会抛异常提示“标准差为0”。我遇到过一列“是否VIP”全是0PySpark直接报错后来在VectorAssembler之前把这一列删掉了。5.4 标准化对稀疏向量的影响如果你在OneHotEncoder之后直接做标准化PySpark的StandardScaler会把稀疏向量变成密集向量因为减均值之后0变成了非0。对于高维稀疏数据这会急剧增加内存占用。我的做法是类别编码后的Vector不做标准化只对连续数值列做标准化然后在VectorAssembler里把两部分拼起来。这样既保留了稀疏性又让连续列尺度一致。6. 特征组装与Pipeline顺序VectorAssembler的“最后一步”原则6.1 Sklearn的ColumnTransformerSklearn用ColumnTransformer把不同列的预处理结果拼在一起from sklearn.compose import ColumnTransformer preprocessor ColumnTransformer( transformers[ (num, StandardScaler(), [age, income]), (cat, OneHotEncoder(), [city]) ] )它自动处理列的对齐和拼接输出一个矩阵。6.2 PySpark的VectorAssemblerPySpark的VectorAssembler把所有输入列拼成一个Vector列输入列必须是数值列或Vector列。它不区分“数值”和“类别”你需要在调用它之前把所有列都准备好。assembler VectorAssembler( inputCols[age_scaled, income_scaled, city_vec], outputColfeatures )这里有个顺序问题VectorAssembler必须放在所有单列变换之后、模型训练之前。如果你在VectorAssembler之后还想对某一列做变换那就得把Vector拆开非常麻烦。所以我的Pipeline顺序固定是缺失值填充 → 类别编码 → 数值标准化 →VectorAssembler→ 模型。6.3 Pipeline的fit与transform分离PySpark的Pipeline和Sklearn的Pipeline都有一个重要特性fit阶段计算所有阶段的参数均值、标准差、类别索引映射等transform阶段应用这些参数。这意味着你必须在训练集上fit然后在训练集和测试集上分别transform。如果直接在测试集上fit就会发生数据泄露。pipeline Pipeline(stages[imputer, indexer, encoder, assembler, scaler]) train_df pipeline.fit(train_df).transform(train_df) test_df pipeline.fit(train_df).transform(test_df) # 错误应该用同一个model正确的写法是model pipeline.fit(train_df) train_df model.transform(train_df) test_df model.transform(test_df)这个错误在Sklearn里也常见但PySpark的Pipeline不会自动帮你检查它只是按顺序执行。6.4 一个真实的踩坑记录我曾经在一个项目里把StringIndexer放在VectorAssembler之后结果StringIndexer报错说输入列不是字符串类型因为VectorAssembler已经把字符串列转成了Vector。排查了半天才发现是阶段顺序写反了。PySpark的Pipeline不会在fit之前校验阶段之间的类型兼容性它只在运行时按顺序执行所以顺序错了就会在某个阶段抛异常。我的经验是在写Pipeline之前先手动把每个阶段单独跑一遍确认输入输出类型对得上再组装成Pipeline。7. 自定义变换UDF与FunctionTransformer的代价对比7.1 Sklearn的FunctionTransformerSklearn允许你用FunctionTransformer把任意Python函数包装成一个变换器from sklearn.preprocessing import FunctionTransformer log_transformer FunctionTransformer(np.log1p, validateTrue)这个函数在pandas的列上逐元素执行速度取决于pandas的向量化程度。7.2 PySpark的UDFPySpark里对应的做法是udffrom pyspark.sql.functions import udf from pyspark.sql.types import DoubleType udf(DoubleType()) def log1p(x): return math.log1p(x) if x is not None else None df df.withColumn(income_log, log1p(col(income)))UDF的代价是它需要在JVM和Python进程之间序列化数据性能比原生Spark函数差一个数量级。如果只是做log1pPySpark有内置的log1p函数from pyspark.sql.functions import log1p df df.withColumn(income_log, log1p(col(income)))内置函数是Catalyst优化的速度比UDF快很多。我只有在内置函数确实做不到的时候才用UDF比如复杂的字符串解析或条件逻辑。7.3 Pandas UDF的折中方案PySpark 2.3之后引入了Pandas UDF它用Arrow做数据传输性能比普通UDF好很多from pyspark.sql.functions import pandas_udf import pandas as pd pandas_udf(DoubleType()) def log1p_pandas(x: pd.Series) - pd.Series: return np.log1p(x) df df.withColumn(income_log, log1p_pandas(col(income)))Pandas UDF的输入输出都是pandas Series适合做向量化操作。但它的启动开销比内置函数大对于简单变换还是优先用内置函数。8. 持续更新中的经验沉淀与版本兼容提醒8.1 两套框架的API演进方向Sklearn的API相对稳定fit/transform/fit_transform的接口多年没变。PySpark的ML模块从1.x到3.x改了不少比如OneHotEncoder在2.3之前不支持dropLastImputer在2.4之前不支持median策略。如果你在维护一个跨版本的代码库建议在代码里加版本判断import pyspark if pyspark.__version__ 3.0: from pyspark.ml.functions import vector_to_array else: # 手动实现 pass8.2 我个人的对照检查清单每次做Sklearn到PySpark的迁移我都会按这个清单过一遍检查项Sklearn行为PySpark行为是否一致缺失值识别np.nannull需转换均值填充覆盖原列生成新列需调整字符串编码OneHotEncoder直接支持需先StringIndexer需两步标准化默认减均值除标准差只除标准差需设withMeanTrue常数列处理缩放到0抛异常需预处理输出类型numpy数组Vector需转换这个清单帮我省了很多调试时间。每次遇到两边结果对不上我就逐项检查通常问题就出在某一项的默认行为差异上。8.3 关于“持续更新”的实践建议这个标题里带了“持续更新中”说明作者打算长期维护这份对照。我的建议是不要只记录“怎么写”还要记录“为什么这么写”和“什么情况下会出错”。比如StringIndexer的handleInvalid参数光写“有三个取值”不够还要写“训练集和测试集类别不一致时用keep但keep会让OneHotEncoder多出一维需要在模型评估时注意”。这些细节才是对照表真正的价值所在。另外版本升级时一定要重新跑一遍对照测试。PySpark 3.0到3.1之间VectorAssembler对null的处理就有变化之前handleInvalidskip会跳过整行后来改成只跳过含null的列。如果不重新测试线上任务可能突然报错。8.4 一个实用的调试技巧当你发现PySpark和Sklearn的结果对不上时不要一上来就怀疑算法。先把数据量降到100行两边分别跑把中间结果打印出来对比。我通常会在PySpark里用df.select(features).show(truncateFalse)看Vector的内容在Sklearn里用print(scaler.mean_)看均值。对比均值、标准差、类别索引这些中间量很快就能定位到是哪一步出了问题。这个方法比逐行读代码快得多。提示PySpark的Vector在show时默认只显示前几个元素如果维度很高可以用vector_to_array转成数组再打印或者用df.select(features).rdd.map(lambda r: r[0].toArray()).take(5)看前5行的完整向量。最后再分享一个小经验如果你在本地用Sklearn做原型然后迁移到PySpark做生产建议把Sklearn的预处理结果保存成Parquet文件然后在PySpark里读进来做对照。这样你可以确保两边输入完全一致排除数据读取阶段的差异。Parquet的列式存储对两边都友好Sklearn可以用pd.read_parquetPySpark可以直接spark.read.parquet。这个做法我在三个项目里用过每次都能快速定位到是预处理逻辑的问题还是数据本身的问题。