SparkML工业级数据流水线：构建可观测、可回滚、可演进的机器学习工程体系

1. 项目概述：这不是“跑个Spark任务”，而是构建可演进的数据智能流水线

“Big-Data Pipelines with SparkML”——光看标题，很多人第一反应是“哦，用Spark MLlib做机器学习”。但干过三年以上数据平台建设的同行都清楚，这五个词背后压着的是整条数据价值链的承重墙：不是模型训练本身，而是让模型能稳定、可信、可持续地嵌入业务决策闭环的工程化能力。我带团队落地过17个跨行业SparkML流水线项目，从金融反欺诈的实时评分到制造设备预测性维护，最常被低估的从来不是算法准确率，而是Pipeline的可观测性、版本一致性、特征复用效率和故障恢复速度。这个标题里，“Pipelines”是主语，“Big-Data”是约束条件，“SparkML”是工具选型——它意味着你必须在分布式、高吞吐、容错优先的环境下，把数据清洗、特征工程、模型训练、评估、部署、监控全链路串成一条“活”的流水线，而不是一堆孤立的notebook脚本。适合谁？不是刚学完《Spark权威指南》的初学者，而是已经写过500行以上DataFrame操作、被生产环境OOM和血缘断裂坑过至少三次的中级数据工程师；也不是只管调参的算法研究员，而是需要和SRE、业务方、合规团队对齐SLA的MLOps实践者。它解决的核心问题，是让机器学习从“实验室里的漂亮指标”变成“每天凌晨三点自动触发、影响千万订单分单策略的可靠服务”。接下来所有内容，都基于一个真实前提：我们讨论的不是如何用SparkML写一个逻辑回归，而是如何让这个逻辑回归在PB级日志中持续产出偏差<0.3%的预测，并在特征源变更时2小时内完成全链路验证与灰度发布。

2. 整体架构设计与核心思路拆解：为什么必须放弃“单Job思维”

2.1 传统误区：把Pipeline当成“训练+保存”的两步操作

很多团队第一次尝试SparkML Pipeline时，会写出这样的代码：读取Hive表→清洗→特征转换→训练→保存Model→用model.transform()跑预测。表面看流程完整，但上线后立刻暴雷：

• 特征不一致：训练时用StringIndexer处理城市字段，预测时新来“海口市”未在训练集出现，直接抛IllegalArgumentException；
• 血缘断裂：业务方问“为什么昨天推荐点击率下降了？”，你翻遍代码才发现两周前有人悄悄改了上游ETL的日期过滤逻辑，但Pipeline没做Schema校验；
• 不可回滚：模型A上线后发现线上AUC掉点，想切回模型B，却发现B的特征处理代码已被覆盖，连训练数据都找不全。

这些问题的根源，在于把SparkML Pipeline当成了“模型序列化工具”，而忽略了它本质是一个声明式的数据处理契约（Contract）。SparkML的Pipeline类不是魔法，它只是把Transformer（如StandardScaler）和Estimator（如LogisticRegression）按顺序封装，关键在于：每个Stage都必须是状态无关、可重复执行、输入输出Schema可验证的确定性组件。

2.2 正确架构：三层解耦的工业级流水线

我们最终采用的架构，是经过6个生产环境迭代验证的“三层解耦”模型：

这个架构的底层逻辑，是把“数据流动”和“模型生命周期”彻底分离。比如特征工程层，我们要求所有特征必须通过FeatureSpec定义：

# 示例：用户行为特征规范（非伪代码，是真实生产代码） from feast import FeatureView, Entity, ValueType user_entity = Entity(name="user_id", value_type=ValueType.STRING) user_behavior_fv = FeatureView( name="user_behavior_features", entities=["user_id"], ttl=timedelta(days=30), schema=[ Field(name="avg_click_per_session_7d", dtype=Float32), Field(name="is_high_value_user", dtype=Bool), # 这个布尔值由规则引擎生成，非模型预测 ], online=True, offline=True, source=user_behavior_source, # 指向Delta表 )

看到这里你可能疑惑：Feast不是Python库吗？怎么和SparkML联动？答案是：SparkML Pipeline只负责“模型部分”，特征由Feature Store统一供给，Pipeline的输入Schema必须严格匹配Feature View的输出Schema。我们在Pipeline构建时，会动态加载Feature View的Schema并做兼容性校验：

# 生产环境强制校验逻辑（已脱敏） def validate_pipeline_input_schema(pipeline: Pipeline, feature_view: FeatureView): expected_fields = set([f.name for f in feature_view.schema]) actual_fields = set(pipeline.getStages()[0].getInputCol()) # 简化示意 if expected_fields != actual_fields: raise PipelineValidationException( f"Feature mismatch! Expected {expected_fields}, got {actual_fields}" )

这种设计牺牲了“写一个脚本就跑通”的便捷性，但换来的是：当业务方新增一个“用户最近3次购买金额标准差”特征时，只需更新Feature View定义，Pipeline无需修改一行代码——因为它的输入接口（Schema）已被契约锁定。这才是“Big-Data Pipelines”的本质：用接口契约代替硬编码依赖，用版本控制代替人工协调。

2.3 为什么坚持用SparkML而非PySpark UDF或自研框架？

当前社区有大量替代方案：用Dask做分布式训练、用Ray on Spark、甚至用Kubeflow Pipelines编排。但我们坚持SparkML，理由很务实：

• 血缘追踪的原生支持：Spark 3.4+ 的explain()可输出完整的逻辑执行计划（Logical Plan），结合Delta Lake的DESCRIBE HISTORY，能精确追溯某次预测结果对应的训练数据版本、特征计算SQL、甚至JVM参数；
• 序列化可靠性：SparkML的PipelineModel.save()生成的是纯Java对象序列化文件（非Python pickle），在YARN/K8s混部环境中，避免了Python版本、包冲突导致的加载失败——我们曾因pandas==1.5.3和1.4.4不兼容，导致线上Pipeline加载超时，而SparkML模型文件无此问题；
• 与现有数仓无缝集成：90%的客户已有Hive Metastore或Delta表，SparkML可直接读取spark.read.table("feature_db.user_features")，无需额外开发适配器。

当然，它也有短板：对深度学习支持弱、超参搜索不如Optuna灵活。我们的应对策略是“分层选型”——SparkML专精于结构化数据的统计模型（GBDT、LR、FM），深度学习模型用TensorFlow Serving独立部署，两者通过Feature Store共享特征。这种混合架构，比强行用一个框架包打天下更符合工程实际。

3. 核心细节解析与实操要点：从代码到生产的12个生死细节

3.1 Pipeline的“不可变性”陷阱：为什么save()后不能修改Stage

SparkML Pipeline的save()方法会将整个Pipeline对象（包括所有Stage的参数）序列化。但很多开发者会犯一个致命错误：

# ❌ 危险操作：保存后修改Stage参数 pipeline = Pipeline(stages=[string_indexer, lr]) model = pipeline.fit(train_df) model.save("hdfs://path/to/pipeline") # 此时string_indexer的handleInvalid="keep" # 后续代码中... string_indexer.setHandleInvalid("keep") # 试图修改，但已无效！

原理：PipelineModel.save()保存的是训练完成后的PipelineModel实例，它内部存储的是Transformer（已拟合）和Estimator（已训练）的快照。string_indexer作为Estimator，其fit()方法返回的是StringIndexerModel（Transformer），而PipelineModel中保存的是这个Transformer，不是原始Estimator。因此，对原始Estimator的修改完全不影响已保存的模型。

正确做法：所有参数必须在fit()前确定，并通过ParamGridBuilder进行超参搜索：

# ✅ 正确：参数网格化搜索 param_grid = ParamGridBuilder() \ .addGrid(string_indexer.handleInvalid, ["keep", "error"]) \ .addGrid(lr.regParam, [0.01, 0.1]) \ .build() cv = CrossValidator(estimator=pipeline, estimatorParamMaps=param_grid, ...) best_model = cv.fit(train_df) # 此时best_model包含最优参数组合 best_model.save("hdfs://path/to/best_pipeline")

提示：我们在线上环境强制要求所有Pipeline必须通过CrossValidator训练，禁用直接fit()。这样既保证参数可追溯，又避免人为误操作。

3.2 特征缩放的“时间维度”灾难：StandardScaler的坑比想象中深

StandardScaler是SparkML中最常用的Transformer，但它的fit()方法默认计算全局均值和标准差。问题来了：如果你的训练数据是“过去30天”，而预测数据是“今天”，那么用30天均值去标准化单日数据，会导致特征分布严重偏移。我们曾在一个电商点击率模型中观察到：工作日的page_view_count均值是120，周末是280，但模型用30天均值180标准化后，周末样本的特征值普遍小于-1，触发了模型对“低活跃用户”的误判。

解决方案：必须实现“时间感知缩放”（Time-Aware Scaling）。我们不使用SparkML内置的StandardScaler，而是自定义TimeWindowedStandardScaler：

class TimeWindowedStandardScaler(StandardScaler): def _fit(self, dataset): # 关键：按时间窗口分组计算统计量 window_spec = Window.partitionBy("date_window").orderBy("timestamp") stats_df = dataset.withColumn( "date_window", F.date_trunc("day", F.col("event_time")) # 按天分窗 ).groupBy("date_window").agg( F.mean("feature_col").alias("mean"), F.stddev("feature_col").alias("std") ) # 将stats_df广播到各Executor，用于transform self._stats_broadcast = self.spark.sparkContext.broadcast( stats_df.rdd.collectAsMap() ) return self def _transform(self, dataset): # transform时，根据event_time查找对应date_window的统计量 date_window = F.date_trunc("day", F.col("event_time")) # 使用broadcast变量做lookup（省略具体join逻辑） return dataset.withColumn("scaled_feature", (F.col("feature_col") - lookup_mean(date_window)) / F.when(lookup_std(date_window) > 0, lookup_std(date_window)).otherwise(1.0) )

这个自定义Transformer的代价是增加了开发复杂度，但换来的是：模型在任意时间窗口的预测稳定性提升37%（A/B测试数据）。记住：在时序敏感场景，任何忽略时间维度的特征工程都是耍流氓。

3.3 模型版本管理的“三权分立”机制

线上模型必须支持灰度发布、AB测试、快速回滚。我们借鉴数据库事务的ACID思想，设计了“三权分立”版本控制：

• 开发权（Dev）：数据科学家在dev分支提交Pipeline代码，触发CI流水线，生成pipeline-dev-20240520-abc123；
• 测试权（Test）：SRE团队将dev版本部署到测试集群，运行全量历史数据回溯（Backtest），生成pipeline-test-20240520-abc123，并输出PSI（Population Stability Index）报告；
• 发布权（Prod）：只有当PSI < 0.1且AUC提升>0.5%时，才允许将test版本Promote为prod，生成pipeline-prod-v1.2.0。

关键实现是用Delta Table管理模型元数据：

-- models_registry表结构（Delta格式） CREATE TABLE IF NOT EXISTS models_registry ( model_name STRING, version STRING, -- 如 v1.2.0 stage STRING, -- 'dev'/'test'/'prod' pipeline_path STRING, -- hdfs://.../pipeline-prod-v1.2.0 train_data_version STRING, -- 对应Delta表的version created_by STRING, created_at TIMESTAMP, psi_score DOUBLE, auc_delta DOUBLE ) USING DELTA;

每次Pipeline训练完成，自动插入一条记录。线上服务通过查询WHERE stage='prod' AND model_name='click_rate'获取最新生产模型路径。这种设计让模型发布从“人肉scp文件”升级为“原子化数据库事务”，发布失败可立即回滚到上一版本。

3.4 在线推理的延迟优化：从秒级到毫秒级的实战技巧

SparkML Pipeline的transform()在批处理场景下性能优秀，但直接用于在线API会遭遇灾难性延迟（平均1.2秒/请求）。我们通过三级优化将其压到85ms以内：

第一级：预热与缓存

• 启动时预加载PipelineModel到内存，并对StringIndexerModel等Transformer做broadcast：

# 预热代码（Flask应用启动时执行） pipeline_model = PipelineModel.load("hdfs://path/to/prod_pipeline") # 广播StringIndexerModel的映射字典（避免每次transform查Driver） indexer_dict = pipeline_model.stages[0].labels # 假设第一个stage是StringIndexer broadcast_dict = spark.sparkContext.broadcast(indexer_dict)

第二级：特征预聚合
不把原始事件流直接喂给Pipeline，而是先用Structured Streaming做5秒窗口聚合：

# 流式特征计算（非Pipeline部分） stream_df = spark.readStream.format("kafka")... \ .withColumn("window_end", F.window(F.col("event_time"), "5 seconds").end) \ .groupBy("user_id", "window_end") \ .agg( F.avg("click_count").alias("avg_click_5s"), F.count("page_view").alias("pv_count_5s") ) # 此时stream_df已是宽表，直接作为Pipeline输入 result_stream = pipeline_model.transform(stream_df)

第三级：JVM调优

• 设置spark.sql.adaptive.enabled=true启用自适应查询执行（AQE）；
• 调整spark.sql.adaptive.skewJoin.enabled=true处理数据倾斜；
• 关键参数：spark.sql.adaptive.localShuffleReader.enabled=true减少网络传输。

实测数据：在16核32G的K8s Pod上，QPS从120提升至890，P99延迟从1240ms降至83ms。记住：在线推理的瓶颈永远不在算法，而在数据搬运和JVM GC。

3.5 数据漂移检测：用Evidently AI填补SparkML的空白

SparkML提供MulticlassClassificationEvaluator等评估器，但仅限于静态数据集。生产环境中，我们需要实时检测“训练数据分布”和“线上数据分布”的差异。我们集成Evidently AI，但做了关键改造：

• 不直接用Evidently的Web Report（太重），而是提取其核心算法：PSI、KS检验、Cramér's V；
• 将检测逻辑嵌入Streaming Query：每10分钟消费一次线上预测日志，计算特征分布并与基准分布对比；
• 触发告警的阈值策略：
  • PSI > 0.25：触发企业微信告警，通知数据工程师；
  • PSI > 0.4：自动暂停该Pipeline的在线服务，切换至兜底规则模型；
  • 连续3次PSI > 0.25：触发自动重训练流水线（AutoML）。

核心代码片段：

def calculate_psi(expected_array, actual_array, n_bins=10): """计算PSI，已优化为向量化计算""" expected_hist, _ = np.histogram(expected_array, bins=n_bins, density=False) actual_hist, _ = np.histogram(actual_array, bins=n_bins, density=False) # 避免除零，加平滑项 expected_pct = (expected_hist + 1e-5) / (len(expected_array) + 1e-5 * n_bins) actual_pct = (actual_hist + 1e-5) / (len(actual_array) + 1e-5 * n_bins) return np.sum((actual_pct - expected_pct) * np.log((actual_pct + 1e-5) / (expected_pct + 1e-5))) # 在Structured Streaming中调用 def drift_detection_udf(expected_dist: list, actual_dist: list) -> float: return calculate_psi(np.array(expected_dist), np.array(actual_dist)) spark.udf.register("psi_calculator", drift_detection_udf, DoubleType())

这套机制让我们在2023年某次上游埋点变更（将user_age从字符串改为整数）中，提前47分钟发现数据漂移，避免了数小时的预测失效。

4. 实操过程与核心环节实现：从零搭建一个可审计的电商点击率Pipeline

4.1 环境准备与依赖管理：为什么我们弃用conda改用Poetry

项目初期，团队用conda管理Python依赖，结果在YARN集群上频繁遇到ModuleNotFoundError: No module named 'pyspark'。根本原因是：conda环境无法被YARN Container正确识别，而Spark on YARN要求所有Python依赖必须打包进--py-files。我们最终切换到Poetry，并制定严格规范：

• pyproject.toml中明确指定pyspark = "^3.4.1"，禁用*通配；
• 构建时执行poetry export -f requirements.txt --without-hashes > requirements.txt，生成无hash的依赖清单；
• 打包命令：zip -r deps.zip $(cat requirements.txt | xargs -I {} pip show {} | grep "Location:" | cut -d' ' -f2 | xargs)；
• 提交作业：spark-submit --py-files deps.zip --files conf/spark-defaults.conf ...。

注意：Spark 3.4.1要求Scala 2.13，而某些旧版Hadoop（如3.2.1）默认Scala 2.12，必须重新编译Hadoop或降级Spark。我们选择后者，因降级风险可控，而重编译Hadoop需全集群重启。

4.2 数据接入层实现：Debezium + Delta Lake的零信任同步

电商订单库是MySQL，我们用Debezium捕获binlog：

# debezium-connector-mysql.yaml name: mysql-connector config: connector.class: io.debezium.connector.mysql.MySqlConnector database.hostname: mysql-prod.internal database.port: 3306 database.user: debezium database.password: ${file:/etc/kafka/secrets:db_password} database.server.id: "184054" database.server.name: mysql_prod table.include.list: inventory.orders, inventory.users snapshot.mode: initial database.history.kafka.bootstrap.servers: kafka:9092 database.history.kafka.topic: schema-changes.inventory

关键配置database.history.kafka.topic将Schema变更事件写入Kafka，我们用Spark Structured Streaming消费该Topic，自动更新Delta表Schema：

# 自动Schema演化（生产环境已验证） schema_evolution_stream = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "kafka:9092") \ .option("subscribe", "schema-changes.inventory") \ .load() def apply_schema_change(batch_df, batch_id): for row in batch_df.collect(): # 解析Debezium Schema变更事件 if row["op"] == "c": # create delta_table = DeltaTable.forName(spark, "inventory.orders") delta_table.generate("symlink_format_manifest") # 触发Manifest更新 elif row["op"] == "u": # update # 执行ALTER TABLE ADD COLUMN（需权限校验） spark.sql(f"ALTER TABLE inventory.orders ADD COLUMNS ({row['new_column']})") schema_evolution_stream.writeStream \ .foreachBatch(apply_schema_change) \ .start()

这套机制实现了“Schema变更自动同步”，无需DBA手动执行DDL，将Schema不一致导致的Pipeline失败率从12%降至0.3%。

4.3 特征工程层实现：Feast + Delta Lake的混合计算

我们定义两个Feature View：

• user_static_features：来自Hive的用户基础信息（性别、地域、注册时间），TTL=永久；
• user_behavior_features：来自Kafka实时流的用户行为（点击、加购、下单），TTL=30天。

关键实现是离线/实时特征的统一查询：

# Feast OnlineStore配置指向Delta表 online_store = FileOnlineStore( config=FileOnlineStoreConfig( path="/delta/feast/online_store" ) ) # 线上服务查询（毫秒级） entity_df = pd.DataFrame({"user_id": ["U123", "U456"], "event_timestamp": [pd.Timestamp.now()]}) feature_vector = store.get_online_features( entity_df=entity_df, features=[ "user_static_features:gender", "user_behavior_features:avg_click_7d" ] ).to_dict() # 离线训练时，直接读取Delta表（无需Feast SDK） train_df = spark.read.format("delta").load("/delta/feast/offline_store/user_features") # 与标签表join labeled_df = train_df.join(label_df, on="user_id", how="inner")

这种设计让特征计算“一次编写，多处运行”，避免了离线用Spark、在线用Redis导致的特征不一致。

4.4 Pipeline构建与训练：从代码到可审计Artifact的全流程

以下是生产环境真实的Pipeline构建代码（已脱敏）：

from pyspark.ml import Pipeline from pyspark.ml.feature import StringIndexer, VectorAssembler, StandardScaler from pyspark.ml.classification import LogisticRegression from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.sql import functions as F # 1. 定义特征列（必须与Feature View Schema严格一致） feature_cols = [ "user_gender_idx", "user_region_idx", "avg_click_7d", "avg_cart_add_7d", "is_vip_user" ] # 2. 构建Pipeline Stage # StringIndexer必须设置handleInvalid="keep"，否则新类别报错 gender_indexer = StringIndexer( inputCol="user_gender", outputCol="user_gender_idx", handleInvalid="keep" # ⚠️ 强制要求 ) region_indexer = StringIndexer( inputCol="user_region", outputCol="user_region_idx", handleInvalid="keep" ) assembler = VectorAssembler( inputCols=feature_cols, outputCol="features" ) scaler = StandardScaler( inputCol="features", outputCol="scaled_features", withStd=True, withMean=True ) lr = LogisticRegression( featuresCol="scaled_features", labelCol="label", predictionCol="prediction", probabilityCol="probability", rawPredictionCol="rawPrediction", regParam=0.01, maxIter=100 ) # 3. 组装Pipeline pipeline = Pipeline(stages=[ gender_indexer, region_indexer, assembler, scaler, lr ]) # 4. 训练与评估（含数据质量检查） train_df = spark.read.format("delta").load("/delta/datasets/train_click") # 强制Schema校验 assert set(train_df.columns) >= set(["user_gender", "user_region", "label"]), "Missing required columns" # 训练 model = pipeline.fit(train_df) # 评估 evaluator = BinaryClassificationEvaluator( labelCol="label", rawPredictionCol="rawPrediction", metricName="areaUnderROC" ) auc = evaluator.evaluate(model.transform(train_df)) print(f"Training AUC: {auc}") # 5. 保存（带元数据） model_path = f"hdfs://namenode:8020/models/click_rate/v1.3.0_{int(time.time())}" model.save(model_path) # 6. 写入模型注册表（Delta） model_registry_df = spark.createDataFrame([{ "model_name": "click_rate", "version": "v1.3.0", "pipeline_path": model_path, "train_data_version": "delta_version_12345", "auc": auc, "created_by": "data_engineer_team", "created_at": F.current_timestamp() }]) model_registry_df.write.format("delta").mode("append").save("/delta/models_registry")

这段代码的关键在于：所有检查（Schema、参数、评估）都是硬编码在训练脚本中，而非靠文档约定。它确保了每次训练产出的Artifact（模型文件+注册表记录）都是可审计、可追溯的。

4.5 线上服务部署：Spark Structured Streaming + Flask的轻量级方案

我们不采用MLflow Model Serving（太重），而是用Flask暴露REST API，后端用SparkSession做批处理：

# app.py from flask import Flask, request, jsonify from pyspark.sql import SparkSession import pandas as pd app = Flask(__name__) # 全局SparkSession（单例） spark = SparkSession.builder \ .appName("click-rate-inference") \ .config("spark.sql.adaptive.enabled", "true") \ .getOrCreate() # 预加载PipelineModel pipeline_model = None @app.before_first_request def load_model(): global pipeline_model pipeline_model = PipelineModel.load("hdfs://namenode:8020/models/click_rate/v1.3.0_1716234567") @app.route('/predict', methods=['POST']) def predict(): data = request.json # 转为Pandas DataFrame pdf = pd.DataFrame(data["features"]) # 转为Spark DataFrame（注意Schema必须匹配） sdf = spark.createDataFrame(pdf) # 执行Pipeline result_sdf = pipeline_model.transform(sdf) # 收集结果（小批量适用） result_pdf = result_sdf.select("prediction", "probability").toPandas() return jsonify(result_pdf.to_dict(orient="records")) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

部署时，用Docker打包：

FROM amazon/aws-cli COPY requirements.txt . RUN pip install -r requirements.txt COPY app.py . CMD ["flask", "run", "--host=0.0.0.0:5000"]

K8s配置限制内存为4G，CPU为2核，实测单Pod可支撑300 QPS。这种方案的优势是：完全复用Spark生态，无需学习新框架，运维成本极低。

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

5.1 “java.lang.OutOfMemoryError: Java heap space” —— 不是内存不够，是序列化爆炸

现象：Pipeline训练到fit()阶段，Executor频繁OOM，但spark.executor.memory已设为16G。

根因分析：SparkML的PipelineModel.save()会序列化整个模型对象。如果Pipeline中包含VectorAssembler且输入列过多（如200+特征），VectorAssembler内部会生成巨大的String数组（存储列名），导致序列化后体积激增。我们曾有一个Pipeline含312个特征，序列化文件达2.1GB，远超Executor堆内存。

解决方案：

• 列裁剪：训练前用CorrelationFilter剔除低相关性特征（corr < 0.05）；
• 分阶段组装：不用单个VectorAssembler，改用多个VectorAssembler分组组装，再用VectorAssembler合并向量：

# 分组组装（降低单个Assembler的列数） assembler_group1 = VectorAssembler(inputCols=group1_cols, outputCol="vec_group1") assembler_group2 = VectorAssembler(inputCols=group2_cols, outputCol="vec_group2") # 合并向量 vector_merger = VectorAssembler(inputCols=["vec_group1", "vec_group2"], outputCol="features")

• 启用Kryo序列化：在spark-defaults.conf中添加：

spark.serializer org.apache.spark.serializer.KryoSerializer spark.kryo.registrationRequired true # 注册SparkML类 spark.kryo.classesToRegister org.apache.spark.ml.PipelineModel,org.apache.spark.ml.feature.StringIndexerModel

实测效果：序列化文件从2.1GB降至87MB，OOM消失。

5.2 “org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException” —— Hive Metastore的隐形依赖

现象：本地IDE运行正常，提交到YARN后报错找不到数据库。

根因：SparkSession创建时，若未显式指定hive.metastore.uris，会默认连接本地Derby数据库。而YARN集群的spark-defaults.conf中配置了spark.sql.hive.metastore.uris指向远程HiveServer2，但PipelineModel.load()内部会创建新的SparkSession，未继承该配置。

解决方案：

• 强制复用当前SparkSession：在load()前，设置系统属性：

import os os.environ['SPARK_HOME'] = '/opt/spark' # 或在submit时添加 --conf "spark.sql.hive.metastore.uris=thrift://hive-server:9083"

• 最佳实践：所有Pipeline操作必须在同一个SparkSession中完成，禁止在Pipeline中新建SparkSession。

实操心得：我们编写了一个PipelineLoader工具类，封装了所有load逻辑，并自动注入当前SparkSession配置，团队新人再也不用踩这个坑。

5.3 “The number of features is different” —— 特征数量不匹配的静默失败

现象：Pipeline训练成功，但线上预测时transform()返回空结果，无报错。

根因：VectorAssembler的inputCols参数是List[String]，如果上游特征计算中某个字段为NULL，VectorAssembler会跳过该列，导致向量维度减少。例如，期望10维，实际只有9维，LogisticRegressionModel内部会因维度不匹配返回空。

排查技巧：

• 训练后立即验证：在model.transform(train_df)后，添加断言：

sample_result = model.transform(train_df.limit(1)) assert sample_result.select("features").first()["features"].size == len(feature_cols), \ f"Feature size mismatch: expected {len(feature_cols)}, got {sample_result.select('features').first()['features'].size}"

• 线上监控：在Streaming Query中，对每批次输出的features向量做size()统计，异常时告警。

5.4 “No space left on device” —— Spark临时目录的磁盘耗尽

现象：Pipeline运行到CrossValidator的fit()时，Executor磁盘爆满。

根因：CrossValidator会并行训练多个Pipeline，每个Pipeline的fit()都会在spark.local.dir（默认/tmp）生成临时文件。而/tmp分区通常只有几GB。

解决方案：

• 修改临时目录：在spark-defaults.conf中：

spark.local.dir /data/spark-temp,/data2/spark-temp

• 清理策略：在YARN NodeManager配置中，添加：

yarn.nodemanager.local-dirs /data/yarn-local,/data2/yarn-local yarn.nodemanager.delete-interval-ms 300000 # 5分钟清理一次

• 终极方案：禁用CrossValidator，改用TrainValidationSplit，它只训练一次，用验证集评估，虽牺牲搜索精度，但稳定性提升。

5.5 “PipelineModel not found” —— HDFS权限与路径的迷雾

现象：模型路径hdfs://namenode:8020/models/v1.0在HDFS中存在，但PipelineModel.load()报FileNotFoundException。

排查步骤：

1. 检查HDFS用户：hdfs dfs -ls /models/v1.0，确认文件属主是spark用户；
2. 检查路径协议：hdfs://vsviewfs://，集群若启用了ViewFS，必须用viewfs://；
3. 检查Kerberos认证：若集群开启Kerberos，需在spark-submit中添加：

--conf "spark.yarn.principal=spark/_HOST@REALM.COM" \ --conf "spark.yarn.keytab=/etc/security/keytabs/spark.service.keytab"

4. 最隐蔽的坑：HDFS路径末尾的斜杠。hdfs://path/和hdfs://path在某些Hadoop版本中被视为