1. 项目概述：用 PySpark MLlib 做分类，不是跑个 demo 就完事了

“Pyspark MLlib | Classification using Pyspark ML”——这个标题看着平平无奇，但我在金融风控建模、电商用户分群、IoT设备异常检测三个不同场景里，用它处理过真实线上日均 2.3TB 的结构化日志数据。它不是教科书里那个调用 LogisticRegression().fit() 就弹出准确率的玩具，而是你得亲手把特征对齐、把稀疏向量喂对、把分区策略调稳、把 OOM 错误压住，最后才能在集群上跑通并稳定产出模型的生产级工具链。核心关键词就三个： PySpark MLlib、分类任务、大规模结构化数据 。它解决的是单机 scikit-learn 完全扛不住的场景——当你的训练集有 5 亿行、42 个特征列（含 17 个类别型变量）、标签分布极度倾斜（正样本仅占 0.37%），你还得在 45 分钟内完成训练+交叉验证+特征重要性输出。适合两类人：一类是刚从 pandas 过渡到 Spark 的数据工程师，另一类是手握海量业务数据却卡在“模型训不出来”的算法同学。别被“MLlib”这个名字骗了，它和 scikit-learn 不是同一套设计哲学：MLlib 是为分布式而生的，它的 API 强制你思考数据分区、序列化开销、宽依赖窄依赖；它的评估器（Estimator）和转换器（Transformer）必须串成 Pipeline 才能复用；它的向量类型（Vector）不支持直接索引，你得用 VectorAssembler 和 StringIndexer 一层层搭积木。我见过太多人把本地调试好的逻辑直接扔进集群，结果卡在 collect() 上等了 2 小时，或者因为没设 setCheckpointDir 导致迭代算法反复 shuffle。这篇不是 API 文档搬运，是我踩过 17 次 java.lang.OutOfMemoryError: GC overhead limit exceeded 、重写过 4 版本特征工程 pipeline、最终把单次训练耗时从 3 小时压到 38 分钟后，整理出来的硬核实操笔记。

2. 整体设计与思路拆解：为什么非得用 MLlib 做分类？又为什么不能照搬 sklearn 思路？

2.1 场景倒逼架构：当数据量突破单机内存天花板

先说一个真实案例：某银行信用卡中心要构建实时反欺诈模型，原始交易日志存于 Hive 表，每天新增 1.2 亿条记录，字段包括 user_id , merchant_id , amount , time_diff_to_last , is_weekend , device_type 等共 38 列。历史训练窗口设为 90 天，总数据量约 108 亿行。用 pandas 读取单日数据已需 42GB 内存，更别说做 One-Hot 编码—— merchant_id 有 2300 万个唯一值，One-Hot 后特征维度将超 2300 万，单机内存直接爆掉。这时候 sklearn 的 LogisticRegression 或 RandomForestClassifier 已经不是“慢”的问题，而是根本无法加载数据。PySpark MLlib 的价值就在这里：它把数据切片（partition）后分散到集群各 executor 上，每个节点只处理自己那份子集，特征工程（如 StringIndexer ）和模型训练（如 GBTClassifier ）都天然支持分布式执行。关键不是“能跑”，而是“能控”——你能精确控制每个 stage 的 shuffle 数据量、指定广播小表、设置 checkpoint 避免血缘过长。这背后是 RDD/DataFrame 的不可变性设计：每次 transform 都生成新 lineage，而 MLlib 的 Pipeline 就是把这一连串不可变操作固化下来，保证训练和预测流程完全一致。

2.2 API 设计哲学差异：Estimator/Transformer/Pipeline 不是语法糖，是生产必需

很多人初学时困惑：“为啥不能像 sklearn 那样 fit(X, y) 一步到位？” 因为 MLlib 的 Estimator （如 LogisticRegression ）本质是一个“训练动作模板”，它不保存数据，只保存超参； Transformer （如 StringIndexerModel ）才是真正的“模型实例”，它保存了 fit 过程中生成的映射字典（比如 merchant_id → index 的哈希表）。这种分离强制你思考两个关键问题：第一，特征工程的可复现性——训练时用 StringIndexer 对 merchant_id 编码，预测时必须用同一个 StringIndexerModel ，否则新来的 merchant_id 会报错 Index out of range ；第二，Pipeline 的原子性——把 StringIndexer 、 VectorAssembler 、 LogisticRegression 串成 Pipeline 后， pipeline.fit(train_df) 返回的是一个完整的 PipelineModel ，它内部已固化所有中间模型， pipelineModel.transform(test_df) 能自动按序执行全部步骤。这避免了 sklearn 中常见的“训练时用 LabelEncoder，预测时忘记 fit 或用错对象”的低级错误。我曾在线上环境修复过一个事故：算法同学本地用 pandas.get_dummies() 做 One-Hot，导出模型后，运维用 Spark SQL 加载新数据时因列名顺序不一致导致预测全错。换成 MLlib Pipeline 后，这个问题从根源上消失。

2.3 分类器选型逻辑：不是参数越多越好，而是要匹配数据分布与业务约束

MLlib 提供的分类器有 LogisticRegression 、 DecisionTreeClassifier 、 RandomForestClassifier 、 GBTClassifier 、 NaiveBayes 五种主流。选哪个？看三个硬指标：
第一，数据规模与稀疏性 。 LogisticRegression 在大规模稀疏数据上收敛快（用 LBFGS 优化器），但要求特征已归一化； NaiveBayes 对高维离散特征友好（如文本 TF-IDF），但假设特征独立，在金融风控中常因“收入”和“房产”强相关而失效。
第二，业务可解释性需求 。风控模型必须给出拒贷理由， DecisionTreeClassifier 的 toDebugString() 可直接输出决策路径，而 GBTClassifier 虽精度更高，但需用 featureImportances + SHAP 近似解释，复杂度陡增。
第三，线上服务延迟要求 。 LogisticRegression 单次预测耗时 < 1ms，适合实时接口； RandomForestClassifier 需遍历上百棵树，P99 延迟常超 15ms，更适合离线批量评分。
我们最终在反欺诈项目中选了 GBTClassifier ，因为 AUC 提升 3.2 个百分点带来的坏账减少，远超延迟增加的成本。但必须强调： maxIter=100 不是拍脑袋定的，而是通过 ParamGridBuilder 在 3 折交叉验证中网格搜索 maxIter=[20,50,100] 、 stepSize=[0.05,0.1,0.2] 得出的最优组合——这部分计算量巨大，必须用 CrossValidator 而非手动循环，否则集群资源浪费严重。

3. 核心细节解析与实操要点：从数据加载到特征工程的 7 个生死关

3.1 数据加载：Hive 表 vs Parquet 文件，分区裁剪怎么写才不拖慢？

加载数据看似简单，却是性能瓶颈第一关。常见错误是 spark.read.table("db.fraud_log") 直接读全表。正确做法是强制谓词下推（Predicate Pushdown）：

# 错误：读全表再过滤，浪费 IO 和内存
df = spark.read.table("db.fraud_log").filter("dt >= '2024-01-01'")

# 正确：让 Hive Metastore 提前裁剪分区
df = spark.read.table("db.fraud_log").where("dt >= '2024-01-01'")

原理在于 where() 触发 Catalyst 优化器将过滤条件下推到数据源层，Hive 只扫描符合条件的分区目录。若数据按 dt 分区，且 dt 是字符串类型，务必确保分区名格式与查询条件严格一致（如 dt=2024-01-01 ，不能写成 dt='2024/01/01' ）。更进一步，对大宽表（>100 列），用 select() 显式指定需要的列：

# 只取关键 12 列，避免读取冗余字段
needed_cols = ["user_id", "amount", "time_diff", "is_weekend", 
               "device_type", "merchant_id", "card_type", 
               "ip_country", "trans_hour", "is_first_trans", 
               "label", "dt"]
df = spark.read.table("db.fraud_log").where("dt >= '2024-01-01'").select(needed_cols)

实测显示，对 500 列的表，显式 select 可减少 40% 的 shuffle 数据量。另外，Parquet 比 ORC 在 Spark SQL 中读取更快（因列式存储 + 更优的字典编码），但若上游是 Hive，优先用 Hive 表而非导出 Parquet——省去 ETL 步骤，且 Hive ACID 事务能保证数据一致性。

3.2 类别型变量处理：StringIndexer 的坑比你想象的多

StringIndexer 是处理 merchant_id 、 device_type 等字符串字段的标配，但三个致命坑必须避开：
坑一：未处理 unseen label 。训练时 merchant_id 有 2300 万个值，预测时来了个新商户 ID， StringIndexerModel.transform() 直接抛 java.lang.IllegalArgumentException: Unseen label 。解决方案是启用 setHandleInvalid("keep") ，它会把新值映射到特殊索引 -1 ，后续在 VectorAssembler 中自动处理为 0 向量：

from pyspark.ml.feature import StringIndexer
indexer = StringIndexer(
    inputCol="merchant_id", 
    outputCol="merchant_id_index",
    handleInvalid="keep"  # 关键！
)

坑二：高频值爆炸 。 merchant_id 中 top 10 商户占交易量 65%，但 StringIndexer 默认按字母序编号，导致索引 0~9 被高频商户霸占，稀疏向量中大量 0 出现在低位，影响 LogisticRegression 收敛速度。应改用 StringIndexer 的 stringOrderType="frequencyDesc" （需 Spark 3.4+），让高频值获得小索引：

indexer = StringIndexer(
    inputCol="merchant_id",
    outputCol="merchant_id_index",
    stringOrderType="frequencyDesc",  # 按频次降序编号
    handleInvalid="keep"
)

坑三：空值传播 。 null 值经 StringIndexer 后变成 null ，但 VectorAssembler 无法处理 null 向量。必须在 StringIndexer 前用 na.fill() 填充：

df = df.na.fill({"merchant_id": "UNKNOWN", "device_type": "OTHER"})

3.3 数值型特征标准化：MinMaxScaler 的陷阱与 RobustScaler 的替代方案

MinMaxScaler 常被推荐，但它在生产环境极危险：训练时 amount 最大值是 99999，某天出现一笔 1000 万的异常交易， transform() 时 (x - min) / (max - min) 计算结果 > 1，后续模型可能溢出。更安全的是 StandardScaler （Z-score），但 amount 等金融数据常呈长尾分布，均值和标准差受异常值扭曲。我们的解法是自定义 RobustScaler ：用 approxQuantile 计算 25% 和 75% 分位数，再用 IQR（四分位距）缩放：

# 计算 IQR
q25 = df.approxQuantile("amount", [0.25], 0.01)[0]
q75 = df.approxQuantile("amount", [0.75], 0.01)[0]
iqr = q75 - q25

# 添加缩放列
df = df.withColumn(
    "amount_scaled",
    when(col("amount") < q25, q25)
    .when(col("amount") > q75, q75)
    .otherwise(col("amount"))
).withColumn(
    "amount_robust",
    (col("amount_scaled") - q25) / iqr
)

这样既抑制了异常值影响，又保证了缩放后数值范围可控（通常在 [-1, 3] 内）。实测在反欺诈数据上，用 RobustScaler 替代 MinMaxScaler ， LogisticRegression 的 AUC 提升 0.8 个百分点。

3.4 特征组合与交互：用 VectorAssembler 构建稠密向量的底层逻辑

VectorAssembler 是拼接特征的“胶水”，但它的行为常被误解。它不生成新列，而是把输入列合并为一个 Vector 类型的列。关键点有三：
第一，输入列必须是数值型 。如果你把 StringIndexer 输出的 merchant_id_index （整型）和 amount_robust （浮点）一起传入，没问题；但若混入字符串列，会报 TypeError: Column x is not numeric 。
第二，null 值处理 。 VectorAssembler 遇到任何输入列为 null ，整个向量变为 null 。所以必须在 assembler 前确保所有列已填充：

# 先填充所有数值列
num_cols = ["amount_robust", "time_diff", "trans_hour"]
for col_name in num_cols:
    df = df.na.fill({col_name: 0.0})

第三，稀疏向量优化 。当类别型变量索引值很大（如 merchant_id_index 最大 2300 万）时， VectorAssembler 默认生成稠密向量，内存爆炸。应显式设置 setHandleInvalid("keep") 并用 VectorSizeHint 提示向量大小：

from pyspark.ml.feature import VectorSizeHint
assembler = VectorAssembler(
    inputCols=["merchant_id_index", "device_type_index", "amount_robust"],
    outputCol="features",
    handleInvalid="keep"
)
# 提示向量最大长度，触发稀疏存储
size_hint = VectorSizeHint(
    inputCol="features",
    size=23000000  # 设为 merchant_id_index 最大值
)

3.5 标签列处理：二分类 vs 多分类，labelIndexer 怎么设才不翻车？

MLlib 所有分类器要求 label 列是 DoubleType，且值为 0.0, 1.0, 2.0...。但业务数据中 label 常是字符串（"fraud"/"normal"）或整型（1/0）。直接 cast("double") 会出错：字符串 "fraud" 无法转 double。必须用 StringIndexer 统一处理：

label_indexer = StringIndexer(
    inputCol="label", 
    outputCol="label_index",
    handleInvalid="error"  # 标签列不允许 unseen 值！
)

注意 handleInvalid="error" —— 标签列绝不能有未知值，否则模型训练会失败。若原始标签有 "pending" 等中间状态，必须在 label_indexer 前用 replace() 清洗：

df = df.replace({"pending": "normal", "review": "fraud"})

对于多分类（如用户分群的 5 个等级）， StringIndexer 会按字典序编号 "A"→0.0 , "B"→1.0 ...，但业务上 "A" 可能是最高风险，应重排顺序：

# 按业务风险升序排列
risk_order = ["D", "C", "B", "A", "E"]  # D 最低，E 最高
df = df.withColumn(
    "label_ranked",
    when(col("label") == "D", 0)
    .when(col("label") == "C", 1)
    .when(col("label") == "B", 2)
    .when(col("label") == "A", 3)
    .when(col("label") == "E", 4)
)

3.6 训练集/测试集划分：randomSplit 的随机种子必须固定

randomSplit([0.8, 0.2], seed=42) 是基础操作，但 seed 不固定会导致每次划分结果不同，模型评估不可复现。更严重的是，若 seed 未设，Spark 用系统时间戳，集群不同 executor 可能生成不同随机数，导致 train_df 和 test_df 数据分布偏差。必须显式指定：

train_df, test_df = df.randomSplit([0.8, 0.2], seed=12345)

但仅此不够。当数据存在时间序列特性（如交易日志）， randomSplit 会打乱时间顺序，导致用未来数据训练、过去数据测试的“数据穿越”。此时必须用时间切片：

# 按 dt 字段切分，确保训练集时间早于测试集
train_df = df.filter("dt < '2024-03-01'")
test_df = df.filter("dt >= '2024-03-01'")

3.7 模型持久化：save() 和 load() 的路径权限与版本兼容性

模型保存不是 model.save("hdfs://path") 就完事。首先，HDFS 路径必须有写权限，且 spark.sql.warehouse.dir 配置正确。其次，MLlib 模型保存是目录结构，包含 _SUCCESS 文件、 metadata 子目录（存模型元信息）、 data 子目录（存实际参数）。加载时必须指向父目录：

# 正确：指向目录
model.save("hdfs://namenode:8020/models/lr_v1")
loaded_model = LogisticRegressionModel.load("hdfs://namenode:8020/models/lr_v1")

# 错误：指向文件
model.save("hdfs://namenode:8020/models/lr_v1/model")
# 加载会报 java.io.FileNotFoundException

最重要的是版本兼容性：Spark 3.3 保存的模型，不能用 Spark 3.2 加载。我们在线上用 spark.version 校验：

if spark.version != "3.3.2":
    raise RuntimeError(f"Model requires Spark 3.3.2, got {spark.version}")

4. 实操过程与核心环节实现：从 Pipeline 构建到评估指标落地的完整链路

4.1 Pipeline 构建：7 步串联，每步都是生产级刚需

以下是我们反欺诈项目中实际运行的 Pipeline，已脱敏并注释关键设计意图：

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler, RobustScaler
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# 步骤1：清洗空值（生产必备）
cleaner = (df
           .na.fill({"merchant_id": "UNKNOWN", "device_type": "OTHER"})
           .na.fill({"amount": 0.0, "time_diff": 0.0}))

# 步骤2：标签索引（二分类，0=fraud, 1=normal）
label_indexer = StringIndexer(
    inputCol="label", 
    outputCol="label_index",
    handleInvalid="error"
)

# 步骤3：商户ID高频排序索引（解决稀疏性）
merchant_indexer = StringIndexer(
    inputCol="merchant_id",
    outputCol="merchant_id_index",
    stringOrderType="frequencyDesc",
    handleInvalid="keep"
)

# 步骤4：设备类型索引
device_indexer = StringIndexer(
    inputCol="device_type",
    outputCol="device_type_index",
    handleInvalid="keep"
)

# 步骤5：鲁棒缩放（抗异常值）
# （此处省略 IQR 计算代码，见 3.3 节）

# 步骤6：特征向量组装（显式指定所有输入列）
assembler = VectorAssembler(
    inputCols=["merchant_id_index", "device_type_index", 
               "amount_robust", "time_diff", "trans_hour"],
    outputCol="features",
    handleInvalid="keep"
)

# 步骤7：梯度提升树分类器（调参后最优）
gbt = GBTClassifier(
    featuresCol="features",
    labelCol="label_index",
    predictionCol="prediction",
    probabilityCol="probability",
    rawPredictionCol="rawPrediction",
    maxIter=100,
    stepSize=0.1,
    maxDepth=5,
    subsamplingRate=0.8
)

# 串联 Pipeline
pipeline = Pipeline(stages=[
    label_indexer, 
    merchant_indexer, 
    device_indexer,
    assembler, 
    gbt
])

# 训练（耗时约 38 分钟）
model = pipeline.fit(cleaner)

这个 Pipeline 的设计意图非常明确： 所有清洗、索引、缩放步骤都固化在 Pipeline 中，保证训练和预测流程 100% 一致 。没有一步是“临时加的”，比如 merchant_indexer 的 frequencyDesc 是为了解决稀疏向量低位聚集问题； subsampleRate=0.8 是为防止过拟合，实测比 1.0 提升泛化能力 2.1%。

4.2 模型评估：BinaryClassificationEvaluator 的 AUC 计算原理与陷阱

MLlib 的 BinaryClassificationEvaluator 默认计算 AUC（Area Under ROC Curve），其原理是：

1. 对测试集每行，模型输出 probability （双元素向量， [p0, p1] ， p1 是正样本概率）；
2. 提取 p1 作为 score，按 score 降序排列所有样本；
3. 计算 ROC 曲线：横轴是 FPR（False Positive Rate），纵轴是 TPR（True Positive Rate），对每个可能的阈值 t ，统计 p1 >= t 的样本中，正样本占比（TPR）和负样本占比（FPR）；
4. 用梯形法积分求 AUC。

陷阱在于： evaluator.evaluate() 默认用 rawPrediction （logit 值）而非 probability ，导致结果偏差。必须显式指定：

evaluator = BinaryClassificationEvaluator(
    labelCol="label_index",
    rawPredictionCol="rawPrediction",  # 注意！这是默认值
    metricName="areaUnderROC"
)
# 但我们要用 probability，需改用：
from pyspark.sql.functions import col, udf
from pyspark.sql.types import DoubleType
# 提取 probability 的第二个元素（正样本概率）
extract_prob = udf(lambda v: float(v[1]), DoubleType())
test_with_prob = test_df.withColumn("prob_fraud", extract_prob(col("probability")))
evaluator = BinaryClassificationEvaluator(
    labelCol="label_index",
    scoreCol="prob_fraud",  # 关键！指定 score 列
    metricName="areaUnderROC"
)
auc = evaluator.evaluate(test_with_prob)

实测显示，用 rawPrediction 计算的 AUC 比用 probability 低 0.015，虽小但影响模型选型判断。

4.3 特征重要性提取：GBTClassifier 的 featureImportances 解析与可视化

GBTClassifier 训练后， model.stages[-1].featureImportances 返回一个 SparseVector ，其 indices 是重要特征索引， values 是重要性得分。但索引对应的是 VectorAssembler 拼接后的顺序，需反查：

# 获取特征名列表（按 assembler.inputCols 顺序）
feature_names = ["merchant_id_index", "device_type_index", 
                 "amount_robust", "time_diff", "trans_hour"]

# 提取重要性
importances = model.stages[-1].featureImportances

# 转为稠密数组并配对
dense_importance = importances.toArray()
feature_importance_df = spark.createDataFrame(
    [(feature_names[i], float(dense_importance[i])) 
     for i in range(len(feature_names))],
    ["feature", "importance"]
).orderBy(col("importance").desc())

# 保存结果
feature_importance_df.write.mode("overwrite").json("hdfs://path/importance_v1")

结果发现 merchant_id_index 占比 42.3%， amount_robust 占 28.7%，印证了“商户集中度”是反欺诈最核心信号。这个结果直接推动产品团队上线“商户黑名单实时拦截”功能。

4.4 模型预测与结果落地：如何把 prediction 写回 Hive 表？

预测不是终点，结果要回写业务系统。常见错误是 model.transform(test_df).select("user_id", "prediction", "probability") 后直接 write ，但 probability 是 Vector 类型，Hive 不支持。必须展开：

from pyspark.sql.functions import col, udf
from pyspark.sql.types import DoubleType

# 提取概率值
extract_prob = udf(lambda v: float(v[1]), DoubleType())
result_df = model.transform(test_df).select(
    "user_id",
    "label_index",
    "prediction",
    extract_prob(col("probability")).alias("prob_fraud"),
    extract_prob(col("rawPrediction")).alias("logit_score")
)

# 写入 Hive 分区表（按日期）
result_df.write \
    .mode("append") \
    .partitionBy("dt") \
    .saveAsTable("db.fraud_prediction_result")

注意 mode("append") 和 partitionBy("dt") ，确保数据按业务日期分区，下游报表可快速查询。

4.5 资源调优实战：Driver 和 Executor 内存、并行度的黄金配比

最后是压测调优。我们集群配置：1 个 Driver（32G 内存），10 个 Executor（每个 16G 内存，4 核）。初始配置 --driver-memory 8g --executor-memory 8g --executor-cores 2 ，训练耗时 2.1 小时。通过 spark.ui 查看 Stage 页面，发现：

• Stage 3（StringIndexer）Shuffle Write 12GB ，但 executor-memory 仅 8G，频繁 GC；
• Stage 7（GBT train）Task Duration 方差极大 ，部分 task 耗时 8 分钟，其他仅 40 秒，说明数据倾斜。

调优步骤：

1. 增大 Executor 内存 ： --executor-memory 12g ，减少 GC；
2. 增加并行度 ： --conf spark.sql.adaptive.enabled=true 开启自适应查询执行（AQE），AQE 自动合并小 task、动态优化 join 策略；
3. 解决数据倾斜 ：对 merchant_id 添加盐值（salting）：

from pyspark.sql.functions import rand, floor, col
salted_df = df.withColumn(
    "merchant_salt", 
    floor(rand() * 10).cast("int")  # 生成 0~9 的盐值
).withColumn(
    "merchant_salt_id", 
    concat(col("merchant_id"), lit("_"), col("merchant_salt"))
)

然后对 merchant_salt_id 做索引，训练后预测时再按 merchant_id 聚合。最终，单次训练稳定在 38 分钟，P95 任务耗时 < 55 秒。

5. 常见问题与排查技巧实录：那些文档里不会写的血泪教训

5.1 OOM 问题速查表：从日志定位根因的 5 种模式

现象	日志关键词	根因	解决方案
Driver OOM	java.lang.OutOfMemoryError: Java heap space	Driver 收集大量数据（如 collect() 、 count() ）	改用 take(100) 或 show() ，禁用 collect()
Executor OOM	java.lang.OutOfMemoryError: GC overhead limit exceeded	Shuffle 数据过大，Executor 内存不足	增大 --executor-memory ，开启 spark.sql.adaptive.enabled
Broadcast OOM	Broadcast variable ... could not be sent	广播大表（>10MB）	改用 mapJoin 或 broadcast 前压缩，或用 BucketJoin
Python UDF OOM	Python worker failed to connect back	Pandas UDF 返回大数据集	限制 UDF 输出行数，或改用原生 Spark SQL 函数
Checkpoint OOM	Checkpoint directory ... is too large	Checkpoint 目录未清理	设置 spark.checkpoint.dir 到 HDFS，并定期 hadoop fs -rm -r

提示：遇到 OOM 第一时间看 yarn logs -applicationId <app_id> ，搜索 OutOfMemoryError ，定位具体 stage 和 task。

5.2 特征工程失败排查：StringIndexer 和 VectorAssembler 的 3 个静默错误

错误1：StringIndexer 输出列名冲突
现象： transform() 后 DataFrame 出现两列 merchant_id_index ，一列是 IntegerType ，一列是 DoubleType 。
原因： StringIndexer 输出列名与已有列同名，Spark 自动重命名（如 merchant_id_index#123 ），但 VectorAssembler 仍找原名。
解决：显式指定 outputCol ，确保唯一： outputCol="merchant_id_idx" 。

错误2：VectorAssembler 输入列类型不一致
现象： transform() 报 java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double 。
原因：某输入列是字符串（如 device_type 未经过 StringIndexer ），但 VectorAssembler 要求数值型。
解决：用 df.dtypes 检查所有输入列类型，确保全是 double 或 integer 。

错误3：PipelineModel 保存后加载失败
现象： load() 报 java.lang.ClassNotFoundException: org.apache.spark.ml.PipelineModel 。
原因：Spark 版本不一致，或 classpath 缺少 spark-mllib_2.12 jar。
解决：确认集群 Spark 版本与开发环境一致， spark-submit 时加 --jars /path/to/spark-mllib_2.12.jar 。

5.3 模型评估偏差：为什么测试集 AUC 高，线上效果差？

这是最痛的问题。我们曾遇到测试集 AUC 0.92，线上监控 AUC 仅 0.78。排查发现：

• 数据漂移（Data Drift） ：测试集用 2 月数据，线上用 3 月数据， merchant_id 分布变化，新商户占比从 5% 升至 18%；
• 特征延迟（Feature Latency） ： time_diff_to_last 特征依赖实时流计算，但流任务偶发延迟，导致特征值为空，被填为 0，模型误判；
• 标签噪声（Label Noise） ：线上“欺诈”标签由人工复核，漏标率 12%，而测试集标签是历史沉淀的“黄金标准”。

解决方案：

1. 每周用 ks_2samp 检验关键特征分布偏移；
2. 特征工程中加入 is_feature_valid 标志列，线上过滤无效特征样本；
3. 用 LabelSmoothing 在训练时降低噪声标签权重。

5.4 生产部署避坑指南：从模型导出到线上服务的 4 个硬性检查点

1. 检查点路径必须设置 ： spark.sparkContext.setCheckpointDir("hdfs://path/checkpoint") ，否则 GBTClassifier 迭代时血缘过长，OOM 风险极高；
2. 禁用 collect() ：所有 df.collect() 必须替换为 df.take(100) 或 df.show() ，线上脚本加入 assert df.count() < 10000 断言；
3. 模型版本强校验 ：在 PipelineModel.load() 后，检查 model.stages[0].uid 是否匹配预期版本号；
4. 资源隔离 ：用 YARN queue 指定 --queue ml-prod ，避免与 ETL 任务争抢资源。

注意：我们线上用 Airflow 调度训练任务，每次运行前自动执行 hadoop fs -du -s hdfs://path/checkpoint/* | grep -v "0 " 清理过期 checkpoint，防止 HDFS 空间耗尽。

5.5 性能对比实测：MLlib vs sklearn 在 1 亿行数据上的硬碰硬

为验证 MLlib 价值，我们在相同硬件（16 核 64G）上对比：

指标	sklearn (Local)	MLlib (Cluster, 10 exec)
数据加载	12 分钟（OOM）	2.3 分钟
特征工程	不可行（内存爆）	8.7 分钟
模型训练	不可行	3