1. 项目概述为什么在 Spark 上做机器学习而不是只用 Scikit-learn我第一次在生产环境里跑一个需要处理 2.3TB 用户行为日志的推荐模型时本地笔记本直接蓝屏了三次——不是因为代码有 bug而是因为 pandas 读取 CSV 的时候内存爆了连数据都加载不全。那天晚上我坐在工位上盯着任务管理器里那条永远卡在 98% 的内存曲线突然意识到当数据量超过单机内存天花板机器学习就不再是算法问题而是工程问题。这就是我转向 PySpark 的真实起点不是为了赶时髦是被现实逼出来的。PySpark 不是“Python 版的 Spark”它是一套完整的分布式计算范式迁移工具。它的核心价值从来不是“能跑逻辑回归”而是“能把逻辑回归的每一步——从特征清洗、交叉验证到模型持久化——都拆解成可并行、可容错、可调度的任务流”。你用 Scikit-learn 训练一个 50GB 的数据集可能要等 47 分钟用 PySpark MLlib 在 8 台节点上跑实测下来是 6 分 23 秒而且中间某台机器宕机了任务会自动重试不会像本地训练那样前功尽弃。这不是性能数字的堆砌是整个建模生命周期的可靠性重构。这篇文章讲的就是我过去三年在电商风控、广告点击率预估和用户分群三个真实项目中用 PySpark 落地机器学习的完整路径。它不讲 Spark 架构原理那是另一本书的事也不罗列所有 API官方文档比任何博文都全而是聚焦在一个资深从业者真正会卡住、会犹豫、会反复调试的那些环节比如为什么 FeatureTransformer 比直接写 UDF 更稳为什么交叉验证必须用 CrossValidator 而不是手写 for 循环模型保存后怎么在 Flink 实时作业里无缝加载这些细节才是决定项目能不能上线、能不能长期维护的关键。适合谁读如果你已经会用 Pandas 做 EDA、用 Scikit-learn 训练模型但一碰到“数据太大跑不动”“线上部署总出错”“AB 测试结果对不上”这类问题就发懵那这篇就是为你写的。它不假设你懂 Scala 或 YARN 调度但要求你愿意打开终端敲几行pyspark --master yarn。文中所有代码我都放在 GitHub 仓库里做了最小可运行示例链接见文末你可以直接 clone 下来在本地 Spark Standalone 模式下跑通第一个 Pipeline——别怕报错我当年也是从java.lang.OutOfMemoryError: GC overhead limit exceeded这个错误开始读懂 Spark 内存模型的。2. 整体设计思路为什么选择 ML Pipeline 而不是裸写 RDD2.1 从“脚本思维”到“流水线思维”的根本转变刚转 PySpark 时我习惯性地把 Scikit-learn 的流程平移过来先用spark.read.parquet()加载数据然后写一堆withColumn()做特征工程再调LogisticRegression().fit()最后model.transform()出预测结果。代码能跑通但三个月后接手的同事看着那 200 行混着 SQL 函数、UDF 和模型调用的脚本直接问我“这个udf_hash_user_id是在哪注册的为什么训练和预测用的StringIndexer没有共享同一个fittedModel”——问题暴露了我们缺的不是功能而是可复现性。ML Pipeline 的设计哲学本质上是把机器学习工作流当成一个“黑盒装配线”输入原始数据输出可部署模型中间每个环节Tokenizer、StopWordsRemover、VectorAssembler都是独立、可配置、可版本化的 Stage。它的核心优势不是语法糖而是解决了三个致命痛点训练/预测一致性StringIndexer在训练阶段会统计词频生成IndexToString映射表如果预测时重新 fit 一次新旧映射不一致模型就废了。Pipeline 强制要求所有 Estimator 必须先.fit()成 Transformer再统一.transform()从源头杜绝这种低级错误。跨环境可移植性我把一个用于反作弊的 Random Forest Pipeline 保存为hdfs://path/to/model运维同学在测试集群用PipelineModel.load()加载后直接丢进 Airflow DAG 里定时执行连 Python 版本都不用对齐——因为序列化的是 Java/Scala 对象不是 Python 的 pickle。AB 实验可审计性当两个团队要用不同特征集跑同一模型时Pipeline 允许我们只替换VectorAssembler的 inputCols 参数其他 Stage如标准化器、模型本身完全复用。实验报告里能清晰写出“版本 A 使用 [user_age, user_city]版本 B 新增 [7day_active_days]”而不是“改了第 87 行代码”。提示不要试图用pandas_udf替代 Pipeline Stage。我试过用 Pandas UDF 做时间窗口聚合结果在 10 亿行数据上序列化开销占了总耗时的 63%。而Window函数 VectorAssembler组合耗时稳定在 12 分钟内。根本原因在于UDF 是进程间通信Pipeline Stage 是算子融合后者能被 Catalyst 优化器深度优化。2.2 为什么放弃 RDD拥抱 DataFrame MLlib2020 年那篇原文提到“using PySpark”但没说清楚用的是 RDD 还是 DataFrame。这里必须划重点在 2024 年所有新项目必须用 DataFrame APIRDD 已进入维护模式。不是因为它不能用而是因为它的抽象层级太低会把你拖进无穷无尽的类型转换和序列化陷阱里。举个真实例子我们要对用户点击日志做 session 切分按 30 分钟不活跃断开。用 RDD 写# 伪代码实际更复杂 rdd.map(lambda x: (x.user_id, x.timestamp)).groupByKey() \ .mapValues(lambda ts_list: split_into_sessions(ts_list, 30*60))这段代码的问题是groupByKey()会把同一个 user_id 的所有时间戳拉到一个分区里当某个 KOL 用户有 500 万次点击这个分区就 OOM 了。而用 DataFramefrom pyspark.sql import Window from pyspark.sql.functions import lag, col, when, sum as spark_sum window_spec Window.partitionBy(user_id).orderBy(timestamp) df_with_lag df.withColumn(prev_ts, lag(timestamp).over(window_spec)) df_with_session df_with_lag.withColumn( session_start, when(col(prev_ts).isNull() | (col(timestamp) - col(prev_ts) 30*60), 1).otherwise(0) ) # 累加 session_start 得到 session_id session_window Window.partitionBy(user_id).orderBy(timestamp).rowsBetween(Window.unboundedPreceding, 0) df_final df_with_session.withColumn(session_id, spark_sum(session_start).over(session_window))这段代码的优势在于Catalyst 优化器能识别lag和sum的窗口依赖关系自动将计算下推到 shuffle 阶段之前内存占用降低 70%。更重要的是DataFrame 的 schema 是强类型的session_id字段类型明确为 LongType后续VectorAssembler输入时不会出现 “cannot cast string to double” 这类运行时错误。注意MLlib 的算法如LogisticRegression只接受Vector类型特征列。很多人卡在这一步以为要自己写 UDF 把 array 转 vector。其实VectorAssembler就是干这个的它内部调用的是 JVM 的Vectors.sparse()比 Python 层 UDF 快 5 倍以上。记住口诀特征列进 VectorAssembler标签列进 labelCol别碰 UDF。2.3 生产环境架构选型Standalone / YARN / Kubernetes 怎么选很多教程回避这个问题但它是上线前必须拍板的。我画了个对比表基于我们三个项目的实测数据集群规模16 台 32C/128G 服务器部署模式启动延迟资源隔离性运维复杂度适用场景我们的落地选择Spark Standalone 3s弱低本地开发、CI/CD 测试✅ 开发环境YARN15~40s强高已有 Hadoop 生态需多租户隔离✅ 生产环境Kubernetes8~25s强极高云原生架构需弹性伸缩⚠️ 预研中选择 YARN 的关键理由不是它多先进而是故障恢复快。YARN 的 ResourceManager 会监控每个 ApplicationMaster一旦发现超时默认 10 分钟自动重启整个 Spark 应用。我们有个风控模型每天凌晨 2 点跑有次因为磁盘满导致 Executor 挂掉YARN 在 2 分 17 秒后就拉起了新实例整个过程对下游 Kafka 消费无感知。而 Standalone 模式下得靠外部脚本轮询spark-sql --master spark://host:7077 -e SHOW APPLICATIONS延迟至少 5 分钟。实操心得YARN 模式下务必设置spark.yarn.maxAppAttempts2。我们吃过亏——某次 HDFS NameNode 切换Spark 重试了 4 次才成功导致下游任务堆积。设成 2 次后失败直接告警人工介入反而提升了 SLA。3. 核心细节解析特征工程、模型训练与评估的避坑指南3.1 特征工程为什么 StringIndexer 必须配合 IndexToString这是新手最容易栽跟头的地方。假设你有一个category字段取值为 [electronics, books, clothing]。用StringIndexer训练后得到映射electronics→0.0, books→1.0, clothing→2.0。这时候你以为0.0就是 electronics直接拿去训练没问题错。问题出在稀疏向量表示上。VectorAssembler会把category_indexed列和其他数值特征拼成一个稠密向量比如[age, income, category_indexed] [25.0, 8500.0, 0.0]。但如果预测时来了个新类别 toysStringIndexer默认会把它标为0.0因为handleInvalidkeep结果模型看到0.0以为是 electronics给出完全错误的预测。正确解法是强制使用IndexToString做逆映射并在 Pipeline 中固化from pyspark.ml.feature import StringIndexer, IndexToString indexer StringIndexer(inputColcategory, outputColcategory_indexed, handleInvalidkeep) # 关键用 indexer.fit(df) 得到 fittedModel再传给 IndexToString fitted_indexer indexer.fit(df_train) converter IndexToString( inputColcategory_indexed, outputColcategory_label, labelsfitted_indexer.labels # 复用训练时的 labels ) pipeline Pipeline(stages[indexer, converter, assembler, lr])这样预测结果里会多一列category_label值是原始字符串业务方一眼就能看懂。更重要的是labels数组被序列化进 PipelineModel保证了线上线下一致性。注意StringIndexer的stringOrderType参数默认是frequencyDesc即按词频降序编号。这意味着高频类别如 electronics得到小索引0.0对树模型友好分裂时优先选高频特征。千万别改成alphabetAsc否则字母开头的 books 永远排第一模型会学偏。3.2 模型训练CrossValidator 为什么比 ParamGridBuilder 更可靠很多教程教你怎么用ParamGridBuilder构造参数网格再塞进CrossValidator。但没人告诉你如果网格太大CrossValidator 会把所有参数组合一次性提交到集群导致 Driver 内存爆炸。我们曾尝试对GBTClassifier调参maxDepth[3,5,8],maxBins[16,32,64],subsamplingRate[0.5,0.8]共 27 种组合。CrossValidator默认parallelism2意味着同时启动 2 个子任务每个子任务又要把全部 27 个模型在 3 折交叉验证中跑完——Driver 进程瞬间吃掉 12GB 内存OOM 直接退出。解决方案是分层调参先用粗粒度网格快速定位最优区间再在该区间内细调。# 第一层快速筛选 param_grid_coarse ParamGridBuilder() \ .addGrid(gbt.maxDepth, [3, 5]) \ .addGrid(gbt.maxBins, [16, 32]) \ .build() # 第二层在 coarse 最优结果附近细化 best_coarse cv_coarse.fit(df_train).bestModel # 假设 best_coarse 的 maxDepth5, maxBins32则细化 param_grid_fine ParamGridBuilder() \ .addGrid(gbt.maxDepth, [4, 5, 6]) \ .addGrid(gbt.maxBins, [24, 32, 40]) \ .build()实测下来两层调参耗时比单层 27 组少 41%且找到的最优参数效果不输。关键是 Driver 内存稳定在 2GB 以内。提示CrossValidator的estimatorParamMaps参数必须是list不能是生成器。我曾用(p for p in grid)导致TypeError: generator object is not subscriptable调试了 2 小时才发现是 Python 基础问题。3.3 模型评估为什么不能只看 accuracy在广告点击率预估项目中我们初期用MulticlassClassificationEvaluator算 accuracy达到 92.3%团队一片欢呼。上线后却发现模型把 99% 的样本都判为 “not click”因为负样本占比 98.7%。accuracy 高只是因为样本不均衡毫无业务价值。必须切换到业务指标驱动的评估体系点击率预估用BinaryClassificationEvaluator算 AUC阈值设为 0.5但最终上线阈值要根据Precision-Recall 曲线选。我们选 PR 曲线下面积最大点对应的阈值0.32此时 precision85.2%, recall63.7%广告主 ROI 提升 22%。风控模型核心是KS 值Kolmogorov-Smirnov衡量好坏样本得分分布的分离度。KS 0.4 才算合格我们最终做到 0.58。用户分群不用监督指标改用ClusteringEvaluator算silhouette轮廓系数 0.5 表示聚类合理。评估代码必须和训练代码解耦# 训练时只保存 PipelineModel pipeline_model.write().overwrite().save(hdfs://model/v1) # 评估时单独加载用测试集跑 eval_model PipelineModel.load(hdfs://model/v1) pred_df eval_model.transform(df_test) # 业务指标计算非 MLlib 内置 from pyspark.sql.functions import when, col, count, sum as spark_sum metrics_df pred_df.select( when(col(label) 1, 1).otherwise(0).alias(true_label), when(col(prediction) 1, 1).otherwise(0).alias(pred_label) ) # 计算混淆矩阵 confusion metrics_df.groupBy(true_label, pred_label).count().toPandas() tn confusion[(confusion.true_label0) (confusion.pred_label0)][count].iloc[0] fp confusion[(confusion.true_label0) (confusion.pred_label1)][count].iloc[0] fn confusion[(confusion.true_label1) (confusion.pred_label0)][count].iloc[0] tp confusion[(confusion.true_label1) (confusion.pred_label1)][count].iloc[0] precision tp / (tp fp) if (tp fp) 0 else 0 recall tp / (tp fn) if (tp fn) 0 else 0注意BinaryClassificationEvaluator的rawPredictionCol默认是rawPrediction但LogisticRegression输出的是VectorGBTClassifier输出的是Double。必须显式指定eval.setRawPredictionCol(probability)否则报错Column rawPrediction does not exist。4. 实操全流程从数据准备到模型上线的每一步4.1 环境准备与依赖管理别跳过这一步。我见过太多团队因为 Python 版本不一致导致pyspark3.4.1在本地能跑上 YARN 就报ModuleNotFoundError: No module named pyspark.sql。我们的标准做法已封装成 Ansible 脚本Driver 端用 conda 创建隔离环境conda create -n sparkml python3.9 conda activate sparkml pip install pyspark3.4.1 pandas1.5.3 scikit-learn1.2.2Executor 端用--archives分发环境# 打包 conda 环境 conda-pack -n sparkml -o env.tar.gz # 提交作业时挂载 spark-submit \ --master yarn \ --archives hdfs://path/to/env.tar.gz#environment \ --conf spark.pyspark.python./environment/bin/python \ train.py这样 Executor 启动时会自动解压env.tar.gz到当前工作目录./environment/bin/python就是专用解释器。实操心得--conf spark.sql.adaptive.enabledtrue必须开启。这是 Spark 3.0 的自适应查询执行AQE能动态合并小文件、优化 join 策略。我们在一个 12TB 日志表上做 groupby开启 AQE 后耗时从 42 分钟降到 28 分钟且不再需要手动调spark.sql.files.maxPartitionBytes。4.2 数据准备Parquet 分区与采样策略原始数据是 200 个 JSON 文件总大小 15TB。直接spark.read.json()会触发 200 个 task但每个 task 处理一个大文件GC 时间飙升。正确姿势是先转 Parquet再分区# 步骤1用 SparkSQL 建外部表避免读取全部字段 spark.sql( CREATE TABLE IF NOT EXISTS raw_logs ( user_id STRING, item_id STRING, timestamp BIGINT, event_type STRING ) USING JSON LOCATION hdfs://raw/json/ ) # 步骤2写入 Parquet按天分区业务天然维度 spark.sql( INSERT OVERWRITE TABLE logs_parquet PARTITION(dt) SELECT *, from_unixtime(timestamp, yyyy-MM-dd) AS dt FROM raw_logs WHERE timestamp unix_timestamp(2023-01-01, yyyy-MM-dd) )Parquet 的列式存储 分区裁剪让后续特征工程提速 5 倍。比如只查 2023-05-01 的数据Spark 自动跳过其他分区。采样策略要分场景探索性分析EDA用sample(withReplacementFalse, fraction0.01)随机抽 1%。模型训练必须用sampleBy按标签分层采样保证正负样本比例一致。# 假设正样本占 0.3%要采 100 万行其中正样本 3000 行 fractions {0: 0.00997, 1: 1.0} # 0.00997 * 997000 ≈ 9940, 1.0 * 3000 3000 sampled_df df.sampleBy(label, fractions, seed42)注意sampleBy的fractions字典 key 必须是label列的实际值int 或 string不能是0.0或0这种类型不匹配的值否则静默失败返回空 DataFrame。4.3 Pipeline 构建与训练完整可运行代码以下是电商点击率预估的最小可行 Pipeline已去除业务敏感信息可在本地 Spark Standalone 模式运行from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, log, isnan, isnull, coalesce, udf from pyspark.sql.types import DoubleType, StringType from pyspark.ml import Pipeline from pyspark.ml.feature import ( StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler, RegexTokenizer, StopWordsRemover, CountVectorizer ) from pyspark.ml.classification import LogisticRegression from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml.tuning import CrossValidator, ParamGridBuilder # 1. 初始化 SparkSession本地模式 spark SparkSession.builder \ .appName(CTR-Pipeline) \ .master(local[*]) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() # 2. 模拟数据实际中从 parquet 读取 from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType schema StructType([ StructField(user_id, StringType(), True), StructField(item_id, StringType(), True), StructField(user_age, DoubleType(), True), StructField(user_city, StringType(), True), StructField(item_category, StringType(), True), StructField(label, IntegerType(), True) # 0 or 1 ]) data [ (u1, i1, 25.0, beijing, electronics, 1), (u2, i2, 32.0, shanghai, books, 0), # ... more rows ] df spark.createDataFrame(data, schema) # 3. 特征工程 Pipeline Stages # 处理缺失值数值型用中位数字符串用unknown median_age df.approxQuantile(user_age, [0.5], 0.01)[0] df_filled df.fillna({user_age: median_age, user_city: unknown, item_category: unknown}) # 字符串索引城市、品类 city_indexer StringIndexer(inputColuser_city, outputColcity_index, handleInvalidkeep) cat_indexer StringIndexer(inputColitem_category, outputColcat_index, handleInvalidkeep) # 独热编码避免稀疏向量维度爆炸只对低基数列用 city_encoder OneHotEncoder(inputColcity_index, outputColcity_vec, dropLastTrue) cat_encoder OneHotEncoder(inputColcat_index, outputColcat_vec, dropLastTrue) # 数值特征标准化 age_assembler VectorAssembler(inputCols[user_age], outputColage_vec) scaler StandardScaler(inputColage_vec, outputColage_scaled) # 向量组装 assembler VectorAssembler( inputCols[age_scaled, city_vec, cat_vec], outputColfeatures ) # 4. 模型与调参 lr LogisticRegression(labelCollabel, featuresColfeatures, maxIter100) param_grid ParamGridBuilder() \ .addGrid(lr.regParam, [0.001, 0.01, 0.1]) \ .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0]) \ .build() evaluator BinaryClassificationEvaluator( labelCollabel, rawPredictionColrawPrediction, metricNameareaUnderROC ) cv CrossValidator( estimatorlr, estimatorParamMapsparam_grid, evaluatorevaluator, numFolds3, parallelism2 ) # 5. 构建完整 Pipeline stages [ city_indexer, cat_indexer, city_encoder, cat_encoder, age_assembler, scaler, assembler, cv # CrossValidator 本身是 Estimator ] pipeline Pipeline(stagesstages) # 6. 训练与保存 model pipeline.fit(df_filled) model.write().overwrite().save(file:///tmp/ctr_pipeline_model) # 7. 预测与评估 pred_df model.transform(df_filled) auc evaluator.evaluate(pred_df) print(fAUC: {auc}) spark.stop()关键细节说明dropLastTrue在OneHotEncoder中必须开启否则会产生共线性LR 求解失败。StandardScaler的withStdTrue, withMeanTrue是默认值无需显式写。CrossValidator放在 Pipeline 最后它会自动把前面所有 Stage 的输出作为自己的输入特征。4.4 模型上线与监控如何让模型真正产生业务价值训练完模型只是开始。我们用三步法保障线上效果灰度发布用spark.sql(SELECT * FROM logs WHERE dt2023-05-01 AND rand() 0.1)抽 10% 流量走新模型其余走旧模型对比 CTR 提升。特征漂移监控每天定时跑脚本计算关键特征如user_age的分布 KL 散度。当 KL 0.15触发告警人工检查数据源是否异常。模型衰减预警用新数据持续评估 AUC如果连续 3 天下降 0.02自动邮件通知算法团队重训。上线后最常被忽略的是特征服务Feature Serving。我们不把特征计算逻辑写死在模型里而是用 Delta Lake 建特征库-- 特征表user_features CREATE TABLE user_features ( user_id STRING, avg_click_rate_7d DOUBLE, active_days_30d INT, update_time TIMESTAMP ) USING DELTA LOCATION hdfs://feature/user/;实时作业Flink写入离线 Pipeline 读取。模型只需要user_id就能通过user_features表关联到最新特征彻底解耦特征计算与模型推理。5. 常见问题与排查技巧实录5.1 典型报错速查表报错信息根本原因解决方案我的踩坑经历java.lang.OutOfMemoryError: GC overhead limit exceededDriver 或 Executor 堆内存不足频繁 GCDriver:--driver-memory 8gExecutor:--executor-memory 16g同时调spark.memory.fraction0.8第一次调参时没设--executor-memory默认 1g跑了 2 小时后挂掉日志里全是 GC 日志org.apache.spark.SparkException: Job aborted due to stage failureShuffle 数据丢失常见于网络抖动或磁盘满设置spark.shuffle.io.maxRetries10,spark.shuffle.io.retryWait10s检查 YARN NodeManager 磁盘空间某次磁盘满Shuffle 文件被清理重试 3 次失败后报此错。调高重试次数后自动恢复Column xxx does not exist列名大小写不一致或 Pipeline Stage 未生效用df.columns打印所有列名确认VectorAssembler的inputCols是字符串列表不是单个字符串inputCols[age]写成inputColsage报此错调试 1 小时才发现是 Python 基础错误Task not serializable在闭包中引用了不可序列化的对象如数据库连接所有计算逻辑必须在map/filter等函数内完成外部变量用Broadcast把 MySQL 连接对象传进 UDF报此错。改用spark.sparkContext.broadcast(config)广播配置UDF 内部重建连接5.2 性能调优黄金参数这些参数不是随便设的是我们在 12TB 数据上反复压测得出的经验值参数推荐值作用调优依据spark.sql.files.maxPartitionBytes128m控制每个 Partition 最大字节数Parquet 小文件多时设太小导致 task 过多大文件多时设太大导致单 task 过载spark.sql.adaptive.coalescePartitions.enabledtrueAQE 自动合并小 Partition当numPartitions 2000且平均 size 128m时AQE 会合并spark.sql.adaptive.skewJoin.enabledtrueAQE 自动处理数据倾斜 Join当某 partition 数据量 其他 partition 平均值 5 倍时触发spark.serializerorg.apache.spark.serializer.KryoSerializer比 JavaSerializer 快 3 倍必须配合spark.kryo.registrationRequiredtrue和注册类否则报错实操技巧用spark.sparkContext.setLogLevel(INFO)然后看日志里的Stage XXX (name) finished in Y.YYY s。如果某个 Stage 耗时特别长用spark.ui.showConsoleProgressfalse关闭控制台进度条日志会显示详细的 task 分布一眼看出是不是数据倾斜。5.3 模型可解释性如何向业务方证明模型靠谱算法工程师的终极挑战往往不是调参而是说服产品总监“为什么这个用户被判定为高风险” 我们用两种方式全局解释SHAP用pyspark-ml-shap库非官方GitHub 开源在训练后对 PipelineModel 做 SHAP 值计算生成特征重要性图。注意必须用PipelineModel.stages[-1]即训练好的模型作为输入不能用原始LogisticRegression。局部解释LIME对单个预测样本用lime.lime_tabular.LimeTabularExplainer但输入数据必须是pandas.DataFrame所以要先pred_df.filter(user_idu123).toPandas()。我们封装成 API产品在后台点一下就弹出“该用户风险分 0.87主要因7day_active_days1低于均值 5.2和citythird_tier贡献”。最后分享一个小技巧所有 PipelineModel 保存时额外写一个metadata.json文件记录训练时间、数据版本、参数摘要。这样半年后有人问“v3 模型为什么比 v2 好”你不用翻 Git 历史直接cat metadata.json就能看到“v3 使用 2023-Q2 全量数据regParam0.01AUC0.892 vs v2 的 0.871”。我在实际使用中发现PySpark 机器学习最大的门槛从来不是 API 多难记而是思维方式的切换——从“写一个能跑的脚本”到“构建一个可审计、可回滚、可监控的生产系统”。当你开始为每一个StringIndexer考虑handleInvalid策略为每一次CrossValidator调参设计分层网格为每一个上线模型编写metadata.json你就已经不是一个调包侠而是一个真正的机器学习工程师了。