做大数据这一行绕不开一种特殊的项目形态机器全摆在内网机房里出不去网装不了包数据敏感度极高但业务方又明确要求把机器学习模型跑起来。这类场景在军用、涉密科研、高安全等级行业里非常典型业内习惯把它笼统地称为军用大数据。它跟你在自己笔记本上pip install pyspark然后跑个鸢尾花分类完全是两码事。前者要考虑的是断网环境下的依赖怎么来、Spark 集群怎么搭、几百 GB 的特征宽表怎么切分、内存不够时 executor 怎么配、模型训完怎么在隔离环境里做批量打分后者只需要考虑代码能不能跑通。这篇文章想聊的就是 Spark 机器学习在强约束的大数据环境下从集群搭建、ETL、特征工程、模型训练到离线推理的一整条链路我会把踩过的坑、参数怎么算、脚本怎么写都摊开讲。不管你是刚接触 Spark 的在校生还是已经写过几个 Spark 作业但没做过大规模调优的工程师应该都能从里面捞到点能直接抄的东西。1. 先想清楚这种场景为什么非得用 Spark 机器学习1.1 单机方案在什么节点上会崩掉很多人第一次接触机器学习用的是 scikit-learn。数据量在几万行、特征几十列的时候它又快又省事fit一下几秒钟就出结果。问题出在数据规模跨过某个门槛之后。我印象很深的一次业务方给了一份约 4000 万行、120 列的用户行为宽表想做一个二分类的流失预测。我先用 pandas 读了一下光是read_csv就吃了 60 多 GB 内存机器直接开始疯狂 swap半小时没读完。这不是代码写得烂的问题是单机架构的物理上限。pandas 的数据结构是纯内存的一个int64就是 8 字节4000 万行乘以 120 列理论上限就是 38 GB 起步中间做merge、groupby还会产生副本实际内存占用往往是理论值的两到三倍。scikit-learn 的多数算法在fit的时候还会要求把整个特征矩阵转成稠密数组特征一 one-hot 展开维度从 120 涨到几千内存直接爆炸。所以判断要不要上分布式的标准很简单当你的数据处理链路里出现了单机内存装不下的宽表或者单次训练时间超过了业务能容忍的窗口就该考虑 Spark 了。这里说的分布式不只是训练分布式更重要的是数据准备阶段的分布式——实际项目里ETL 和特征工程的时间往往占整个流程的 70% 以上模型训练反而很快。1.2 Spark 相比其他分布式框架的实际取舍分布式计算框架不止 Spark 一家为什么在涉密内网场景里Spark 出现的频率明显更高我自己的观察有几个原因。第一是生态完整度。Spark 自带 Spark SQL、Spark MLlib、Spark Streaming、GraphX一个框架把批处理、SQL 查询、机器学习、流计算全包了。内网环境装东西本来就麻烦能少部署一套就少一套。你用 Flink 做流、用别的做批、再用第三方库做机器学习光是版本兼容就能折腾一周。第二是离线部署友好。Spark 的发行包是一个解压即用的目录不依赖外部服务配上 JDK 就能跑Standalone 模式连 YARN 都不需要。这在完全断网的机房里意义很大——你不需要去编译源码、不需要联网拉一堆依赖一个 tar 包搞定。第三是编程接口统一。Python、Scala、Java、R 都能写团队里有人只会 Python有人习惯 Scala都能上手。MLlib 的 Pipeline API 跟 scikit-learn 的设计思路很像从 sklearn 转过来的人基本半天就能上手。当然它也有短板。MLlib 的算法覆盖面比 sklearn 窄一些前沿模型比如某些图神经网络、复杂的梯度提升实现它没有对超大规模稀疏特征的处理它也不如专门的参数服务器框架灵活。所以我的经验是特征维度在十万级以内、样本量在十亿级以内、算法是常规的树模型或线性模型Spark MLlib 完全够用如果算法特别新那就 Spark 做数据、单机做训练把特征抽样出来喂给小模型。1.3 离线内网带来的三个硬约束这类项目跟公网项目的最大区别不在技术栈而在环境约束。我把它们归纳成三条后面所有章节的方案设计基本都是在回应这三条。注意以下约束在任何强合规的离线环境里都成立做方案设计时要把它们当成前置条件而不是遇到再说。依赖无法在线获取。pip、conda、apt、yum、Maven 全部不可用所有安装包必须提前在能联网的机器上下好用移动介质带进去。Python 生态的麻烦在于依赖树很深一个 pyspark 背后可能牵扯几十个包得用pip download把整棵树的 wheel 都拉下来。数据不能出内网。这意味着你不能把数据上传到任何外部的平台做标注或训练所有计算必须在内网闭环完成。同时也意味着模型评估只能靠本地留出的验证集没有外部基准可比。审计与可追溯。谁在什么时间跑了什么作业、用了哪些数据、输出了什么结果通常都要留痕。这就要求作业命名规范、日志集中收集、产出表统一管理不能像在自己电脑上那样随手df.show()就完事。2. 集群规划核数、内存和存储到底怎么算2.1 节点角色的划分逻辑一个最小可用的 Spark 集群需要主节点Master和工作节点Worker。生产环境里通常还会拆出独立的 NameNode、ResourceManager 节点。我的习惯是按下面这种角色分配10 台机器左右的规模比较典型节点角色数量主要职责配置建议主控节点2一主一备Spark Master、HDFS NameNode、YARN RM32 核 / 128 GB / SSD 系统盘计算节点8Spark Worker、NodeManager、DataNode64 核 / 256 GB / 12 块数据盘边缘节点1提交作业、跑客户端、装开发环境16 核 / 64 GB边缘节点的作用经常被忽略但它其实很关键。不要把spark-submit直接跑在主控节点上客户端进程会吃掉主节点的内存严重时能把 NameNode 拖挂。所有作业统一从边缘节点提交主节点只负责调度。至于主控节点为什么要一主一备是因为 Standalone 模式下 Master 是单点它挂了整个集群就不可用。用 ZooKeeper 做 HA 至少要三台成本高很多涉密项目规模不大用一台备机加手动切换就够了配合监控告警实际影响可控。2.2 executor 内存与核数的推算过程这是整个集群配置里最容易拍脑袋、也最影响性能的部分。我见过太多人直接把spark.executor.memory设成100g然后抱怨作业老是 OOM。内存不是越大越好配置的核心是让每个 executor 的内存占用处在 JVM 能稳定管理的区间。以单台 64 核 256 GB 的计算节点为例推算步骤是这样的第一步确定操作系统和 Hadoop 守护进程要预留多少。NodeManager、DataNode 这些常驻进程加上系统本身一般预留 20% 到 25%。256 GB 取 24 GB 预留剩下 232 GB 给 Spark executor。核数这边留 4 核给系统和守护进程剩 60 核。第二步确定每个 executor 用几个核。经验值是 4 到 5 个核。核数太少每个 executor 的并行度低广播变量和 shuffle 的效率上不去核数太多单个 executor 里十几个任务抢内存GC 压力大还容易因为一个任务慢拖垮整组。取 5 核。第三步算 executor 数量。executor 数 每节点可用核数 / 每 executor 核数 60 / 5 12。第四步算每个 executor 能分到多少内存。232 GB / 12 ≈ 19.3 GB。但要注意Spark 的spark.executor.memory之外还有一块堆外内存spark.executor.memoryOverhead默认是max(executorMemory * 0.1, 384MB)。如果设spark.executor.memory16goverhead 默认约 1.6 GB每 executor 实际占用 17.6 GB12 个就是 211 GB安全。如果设成 18goverhead 1.8 GB总共 19.8 GB 乘以 12 是 237 GB超过了 232 GB 的上限YARN 会直接拒绝。所以最终配置是spark.executor.cores5、spark.executor.memory16g、spark.executor.memoryOverhead2g每台 12 个 executor10 台机器8 计算 2 主控如果不参与计算的话总共约 96 个 executor。这里有个容易翻车的点YARN 的可用内存是按executor.memory memoryOverhead算的不是只算 memory。很多教程只讲 memory导致你按 256/2012 算完实际提交时 YARN 报Required AM memory ... is above the available。把 overhead 一起算进去就不会出错。2.3 存储选型HDFS 还是本地盘数据放哪直接决定了 shuffle 和读写性能。Spark 本身不存储数据它依赖外部存储。离线内网场景里主流选择是 HDFS理由有几点三副本保证数据不丢节点坏了数据还在计算时数据本地性data locality好任务尽量调度到数据所在的节点跟 YARN 配合成熟运维资料多。不过在涉密环境里磁盘容量的规划要留足余量。假设原始数据 50 TBHDFS 三副本就是 150 TB中间结果表、特征宽表、模型产出加起来实际占用往往是原始数据的四到六倍。我的做法是原始层保留三副本中间层和产出层可以降到两副本通过dfs.replication按目录设置能省下相当一部分空间。如果是小规模集群少于 5 台或者数据量在 10 TB 以内也可以用本地磁盘加spark.local.dir多目录挂载的方式省掉 HDFS 的运维成本。但一旦要跨节点 shuffle 大量数据HDFS 的优势就压不住了还是老实上分布式文件系统。3. 完全离线环境下的集群部署实操3.1 依赖包的本地化准备联网机器上的准备工作是整个离线部署里最耗时间、也最容易漏东西的环节。我列一下实际需要带进内网的东西清单# 1. JDKSpark 3.x 建议 JDK 8 或 11 jdk-8u371-linux-x64.tar.gz # 2. Hadoop 发行包如果用 HDFS hadoop-3.3.6.tar.gz # 3. Spark 发行包注意选带 hadoop 的版本 spark-3.5.1-bin-hadoop3.tgz # 4. Python 及其依赖的离线 wheel # 在联网机器上执行把整棵依赖树拉下来 pip download pyspark3.5.1 -d /opt/wheelhouse pip download numpy pandas pyarrow -d /opt/wheelhouse # 5. 如果有 Scala 作业还要带 Maven 本地仓库 # 在家目录的 .m2/repository 里打包整个目录这里有个大坑要提醒pip download默认只下载当前平台的 wheel如果联网机器是 x86_64 而内网机器是 ARM 架构wheel 装不上。跨架构时要用--platform和--python-version显式指定或者干脆在架构一致的机器上准备。另一个坑是离线安装时的依赖顺序。pip 在--no-index模式下一旦遇到缺失依赖就整批失败所以要确认/opt/wheelhouse里包是齐的。可以用pip install --no-index --find-links/opt/wheelhouse pyspark3.5.1先在小机器上试装一遍确认没问题再正式部署。3.2 配置文件的逐项填写Spark 的核心配置集中在$SPARK_HOME/conf下的三个文件。我把生产环境里常用的配置列出来逐条说明为什么这么设。spark-env.shexport JAVA_HOME/usr/local/jdk1.8.0_371 export SPARK_MASTER_HOSTnode01 export SPARK_MASTER_PORT7077 export SPARK_MASTER_WEBUI_PORT8080 export SPARK_WORKER_CORES60 export SPARK_WORKER_MEMORY230g export SPARK_WORKER_INSTANCES1 export SPARK_LOCAL_DIRS/data1/spark,/data2/spark,/data3/spark export HADOOP_CONF_DIR/opt/hadoop/etc/hadoopSPARK_LOCAL_DIRS这一项特别重要。Spark 的 shuffle 中间文件默认写在/tmp数据量一大就把系统盘写满作业直接失败。把它指向多块独立的数据盘是性价比最高的一个调优动作用便宜的 SSD 或 SAS 盘都能显著提升 shuffle 性能。spark-defaults.conf这些是全局默认值spark.master spark://node01:7077 spark.eventLog.enabled true spark.eventLog.dir hdfs:///spark-logs spark.eventLog.compress true spark.serializer org.apache.spark.serializer.KryoSerializer spark.sql.shuffle.partitions 600 spark.memory.fraction 0.6 spark.memory.storageFraction 0.5 spark.sql.adaptive.enabled true spark.sql.adaptive.coalescePartitions.enabled true spark.driver.memory 8g解释几个关键项。spark.serializer设成 Kryo 而不是默认的 Java 序列化是因为在大规模数据下Kryo 的序列化速度和体积通常能优化 30% 到 50%对 shuffle 密集型的作业提升明显。但 Kryo 有个要求自定义的类需要注册否则还是走通用序列化优化打折。用sparkConf.registerKryoClasses(Array(classOf[MyClass]))显式注册。spark.sql.shuffle.partitions默认 200这个值在数据量小时没问题数据一大就成瓶颈。假设一次 shuffle 的中间数据是 300 GB每个分区建议控制在 100 到 200 MB分区数 300 * 1024 / 128 ≈ 2400。设 600 是折中——分区太多会带来大量小文件和小任务调度开销太少又会 OOM。开了 AQE自适应查询执行之后Spark 会在运行时自动合并小分区、拆分大分区所以基准值给个偏大的数就行。spark.memory.fraction0.6表示 60% 的堆内存用于执行和存储剩下 40% 留给对象本身和元数据。如果作业里有大量缓存 RDD 或 DataFrame可以调到 0.7如果 GC 频繁反而要降到 0.5。3.3 启动验证与冒烟测试配置写完按node01到node10的顺序启动# 启动 HDFS $HADOOP_HOME/sbin/start-dfs.sh # 启动 Spark 集群 $SPARK_HOME/sbin/start-all.sh # 确认工作节点都注册上了 $SPARK_HOME/bin/spark-shell --master spark://node01:7077进去之后先做一个最基础的冒烟测试确认集群真能干活// 用 Spark 自带的数据算一下 Pi确认 executor 起来了 val count sc.parallelize(1 to 100000).map { _ val x math.random; val y math.random if (x*x y*y 1) 1 else 0 }.reduce(_ _) println(sPi is roughly ${4.0 * count / 100000})同样别忘了用 Python 环境再验一次$SPARK_HOME/bin/pyspark --master spark://node01:7077rdd sc.parallelize(range(1000), 10) print(rdd.sum())提示冒烟测试通过不代表环境没问题。真正要验的是跨节点通信和数据本地性。建议从 HDFS 读一个几 GB 的文件跑一次count()和一次groupBy().count().collect()看 WebUI 上的任务是不是均匀分布在各节点有没有出现只在一个节点上跑的情况。4. 数据接入与 ETL从原始表到特征宽表4.1 数据源接入与分区策略内网环境里的数据源通常是 Hive 表、HDFS 上的 Parquet/ORC 文件或者是从业务库导出的 CSV。不管哪种接入时的第一件事是确认分区分区字段选得好后续所有读取都能做分区裁剪扫描数据量能降一个数量级。我一般按时间分区dtyyyy-MM-dd。如果数据量特别大单日超过 1 TB再加上业务维度做二级分区比如dt2024-06-01/region01。分区键的选择有讲究分区数不宜超过十万否则 Hive 的 metastore 和 NameNode 都会被拖慢。如果某个维度基数太高比如用户 ID 有几千万千万别拿它做分区改用分桶bucketing或者干脆只在文件内部做排序。读取时的一个实用技巧是用spark.sql.hive.metastore.jars指向内网本地的 Hive jar避免每次启动都去拉起 metastore 连接。在离线环境里连接超时的默认时间太长卡住的时候很难判断是网络问题还是配置问题把这个参数提前设好能省不少排查时间。4.2 ETL 脚本的写法要点ETL 脚本我习惯用 SQL 写主体逻辑Python 做调度和参数化。原因是 SQL 可读性好业务方也能看出问题的时候排查路径清晰。一个典型的特征宽表构建脚本长这样-- 从明细表聚合出最近 90 天的统计特征 INSERT OVERWRITE TABLE dw.feature_wide PARTITION (dt${biz_date}) SELECT t.user_id, t.item_id, COUNT(1) AS cnt_90d, AVG(t.amount) AS avg_amount_90d, STDDEV(t.amount) AS std_amount_90d, MAX(t.event_time) AS last_event_time, SUM(CASE WHEN t.event_type pay THEN 1 ELSE 0 END) AS pay_cnt_90d, -- 趋势斜率用线性回归的简化近似避免引入额外 UDF (MAX(CASE WHEN t.days_ago 15 THEN t.amount END) - MAX(CASE WHEN t.days_ago BETWEEN 76 AND 90 THEN t.amount END)) / 75.0 AS trend_slope FROM dw.event_detail t WHERE t.dt BETWEEN date_sub(${biz_date}, 90) AND ${biz_date} GROUP BY t.user_id, t.item_id ;几个实操要点。第一聚合的粒度要跟最终的样本粒度对齐。如果训练样本是用户-物品对那 ETL 的GROUP BY就必须是这两个字段否则后面还要再做一次 join白白多一次 shuffle。第二趋势类特征尽量用 SQL 里的条件聚合实现实在要用窗口函数也要控制分区大小Window.partitionBy(user_id)在用户量千万级的时候会把所有数据集中到一个分区直接 OOM。第三别在 ETL 阶段做 one-hot 编码那是特征工程该干的事放这里会让中间表膨胀几百倍。4.3 特征工程流水线的构建MLlib 的 Pipeline 设计思路跟 sklearn 几乎一样但有个关键差别它的 Transformer 是延迟执行的只有遇到fit或transform才真正跑。理解这一点对内存管理很重要——你在 Pipeline 里定义十个 stage不代表它会同时占十份内存。一个完整的数值加类别特征处理流水线from pyspark.ml.feature import ( StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler, Imputer ) from pyspark.ml import Pipeline cat_cols [unit_type, region_code, item_class] num_cols [cnt_90d, avg_amount_90d, std_amount_90d, trend_slope] # 类别特征先索引成数字再 one-hot indexers [ StringIndexer(inputColc, outputColc _idx, handleInvalidkeep) for c in cat_cols ] encoders [ OneHotEncoder(inputColc _idx, outputColc _oh) for c in cat_cols ] # 数值特征缺失值填充 imputer Imputer( inputColsnum_cols, outputCols[c _fill for c in num_cols] ).setStrategy(median) # 拼成向量 assembler VectorAssembler( inputCols[c _fill for c in num_cols] [c _oh for c in cat_cols], outputColraw_features, handleInvalidskip ) # 标准化树模型其实不需要但线性模型和神经网络必须 scaler StandardScaler( inputColraw_features, outputColfeatures, withMeanTrue, withStdTrue ) pipeline Pipeline(stagesindexers encoders [imputer, assembler, scaler])handleInvalidkeep这一项要特别注意。默认值是error遇到没见过的类别直接抛异常离线批处理里这会导致整个作业在某一行脏数据上挂掉。生产环境一律设成keep或skipkeep会为未知类别单独分一个索引skip则直接过滤掉这行。选哪个取决于业务能不能容忍丢样本——风控类场景通常不能丢用keep统计分析类场景可以用skip。StringIndexer还有个坑它的索引顺序是按类别出现频率降序排列的每次运行时如果数据分布变了索引映射也会变导致模型上线后特征错位。解决办法是用StringIndexerModel的save把索引器持久化预测时加载同一个模型而不是重新 fit。5. 模型训练算法匹配、调参与内存控制5.1 算法选择与实际场景的匹配MLlib 里的算法不算多但覆盖了绝大多数常规需求。我的选择逻辑是看三件事样本量、特征稀疏性、以及要不要可解释性。场景特点推荐算法理由样本量大、特征稀疏十万级以上线性 SVM / 逻辑回归稀疏向量处理效率高训练快特征稠密、要处理非线性、要特征重要性GBT / 随机森林精度高能输出特征重要性需要概率输出、要快逻辑回归输出校准过的概率解释性强特征维度极高百万级线性模型 特征哈希避免维度爆炸和字典维护成本我用得最多的是 GBT梯度提升树。它在结构化数据上的表现通常比随机森林好一截代价是训练更慢、参数更敏感。GBT 在 MLlib 里只支持二分类和多分类不支持回归的概率输出这点跟 XGBoost 有区别如果你的场景必须要回归用GBTRegressor参数体系基本一致。注意MLlib 的 GBT 是逐棵串行训练的boosting 的本质决定没法像随机森林那样靠numTrees并行。所以maxIter设太大时训练时间会线性增长。生产上先设 50 到 100看验证集指标什么时候收敛再决定要不要加。5.2 交叉验证与参数网格的实操调参这块MLlib 提供了CrossValidator和TrainValidationSplit。前者做 k 折交叉验证稳但慢后者只切一次快但结果波动大。数据量大且资源紧张时优先用TrainValidationSplit因为 k 折意味着要训练 k 倍次数的模型。from pyspark.ml.classification import GBTClassifier from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator gbt GBTClassifier( labelCollabel, featuresColfeatures, maxIter80, seed42 ) grid (ParamGridBuilder() .addGrid(gbt.maxDepth, [4, 6, 8]) .addGrid(gbt.stepSize, [0.05, 0.1]) .addGrid(gbt.maxBins, [32, 64]) .build()) # 3 * 2 * 2 12 组参数 evaluator BinaryClassificationEvaluator( labelCollabel, metricNameareaUnderROC ) cv CrossValidator( estimatorgbt, estimatorParamMapsgrid, evaluatorevaluator, numFolds3, parallelism6, # 同时跑 6 组参数 seed42 ) cv_model cv.fit(train_df)这里parallelism6是个关键参数。它控制同时评估多少组参数设得太小浪费集群资源设得太大又会导致每个模型分到的 executor 太少、还可能因为内存不够排队。经验值是parallelism 集群总 executor 数 / 4左右。96 个 executor 的话设 6 到 12 之间比较合理。内存方面CrossValidator是内存杀手。它会把训练数据缓存在内存里供多轮复用如果数据本身就有几十 GB加上多个模型同时训练很容易 OOM。做法是把训练集先cache()并把storageLevel设成MEMORY_AND_DISK让溢出的部分落盘别指望纯内存扛住。5.3 训练过程中的内存监控与调优判断一个训练作业是不是健康看三个东西GC 时间占比、shuffle spill 的量、以及任务执行时间的离散程度。这三个指标在 Spark WebUI 的 Stage 详情页都能看到。GC 时间占比超过 10% 就要警惕了。原因通常是 executor 内存配得太大JVM 在 Full GC 时停顿时间过长。这时候不要急着加内存反而应该把 executor 内存调小、executor 数量加多让每个 JVM 管理的堆更小。实践里16 GB 到 20 GB 的 executor 内存是个比较舒服的区间超过 32 GB 之后 GC 收益会急剧下降。Shuffle spill磁盘溢写的量要是接近 shuffle read 的量说明内存不够数据大部分落盘了。解决办法有两个方向一是增大spark.sql.shuffle.partitions让每个分区处理的数据更少二是提升spark.memory.fraction给执行内存更多空间。任务执行时间离散度高比如中位数 2 秒、最大值 300 秒基本可以断定是数据倾斜这个下一章专门讲。顺便提一下spark.sql.adaptive.enabledtrue开启 AQE 之后Spark 会自动处理一部分倾斜skewJoin优化但它只对 join 生效对groupBy造成的倾斜无能为力还是得手工处理。6. 数据倾斜分布式训练里最难缠的对手6.1 倾斜在作业里的三种表现数据倾斜的本质是某个 key 的数据量远超其他 key导致处理这个 key 的任务成为长尾。在 Spark 作业里它通常表现为三种形式。第一种是任务时间分布极端不均。WebUI 上看到一个 Stage 里 599 个任务 1 秒跑完剩 1 个跑了 20 分钟。这是最典型的倾斜信号。第二种是单任务 OOM。注意这里的 OOM 不是 executor 级别的而是某个任务处理的数据量太大shuffle read的时候就把它撑爆了。报错信息里能看到FetchFailedException或者 executor 被 YARN kill 掉。第三种是整个 Stage 卡在 99% 不动。因为只有最后一个任务没完成而它在慢慢吞吞地处理那个超大的 key其他资源全闲着。排查的时候有个固定动作先把有倾斜的 DataFrame 按疑似 key 做一次count()按数量倒序看前 20 个。df.groupBy(user_id).count().orderBy(F.desc(count)).show(20, truncateFalse)如果排第一的 key 数量是第二名的十倍以上那就是它了。常见的倾斜 key 有空值null或者、默认值比如unknown、-1、以及少数几个头部用户。空值和默认值最容易处理直接在 ETL 阶段过滤或改写就行但很多人会忽略。6.2 加盐与广播 Join 的具体写法如果是 join 造成的倾斜而且其中一边是小表首选广播 join。把小的那张表广播到所有 executor大表就不用 shuffle 了倾斜自然消失。from pyspark.sql.functions import broadcast result big_df.join(broadcast(small_df), onitem_id, howleft)广播的阈值由spark.sql.autoBroadcastJoinThreshold控制默认 10 MB。如果小表有 50 MB可以调到 100 MB但别调太大——广播变量是每个 executor 存一份50 MB 乘以 100 个 executor 就是 5 GB 的额外内存。如果两边都是大表那就得上加盐salting。思路是给大表的热点 key 随机加后缀把一条记录打散成多条小表那边把每个 key 复制成对应的份数再按加盐后的 key 做 join。import pyspark.sql.functions as F salt 16 # 大表给热点 key 加随机后缀 big_df big_df.withColumn( salt, (F.rand(seed42) * salt).cast(int) ).withColumn( join_key, F.concat_ws(_, F.col(item_id), F.col(salt)) ) # 小表每个 key 复制成 salt 份 small_df small_df.withColumn( salt, F.explode(F.array([F.lit(i) for i in range(salt)])) ).withColumn( join_key, F.concat_ws(_, F.col(item_id), F.col(salt)) ) result big_df.join(small_df, onjoin_key, howleft)这个写法的代价是数据膨胀了salt倍所以salt不能设太大。我一般从 8 或 16 开始试看倾斜是否缓解逐步调整。加盐只对热点 key 有效如果每个 key 都加等于把倾斜均匀放大反而更慢。所以更精细的做法是先识别出前 N 个热点 key只对它们加盐其余走普通 join最后 union 回来。6.3 序列化与 shuffle 参数的配合序列化方式对倾斜的影响经常被忽视。用默认的 Java 序列化时一个热点 key 对应的数据块可能比 Kryo 大两到三倍本来勉强能处理的任务就变成 OOM 了。所以前面说的KryoSerializer不只是性能优化也是降低倾斜风险的一道保险。配合几个 shuffle 层面的参数spark.reducer.maxSizeInFlight 96m spark.shuffle.file.buffer 1m spark.shuffle.io.maxRetries 10 spark.shuffle.io.retryWait 10smaxSizeInFlight控制每个 reducer 一次能拉取多少数据默认 48 MB。在带宽充足的内网环境里调到 96 MB 能减少拉取轮次。shuffle.io.maxRetries调大是因为大规模集群里偶发的节点超时很常见默认 3 次重试在数据量大时经常不够作业会因为一个网络抖动整体失败重试成本极高。7. 常见问题速查与排查思路7.1 报错对照表做了几年下来实际遇到的报错翻来覆去就那么几个。整理成表方便对照。报错关键词典型原因处理动作Container killed by YARN for exceeding memory limitsexecutor 内存或 overhead 不够提高 memoryOverhead 到 memory 的 15%java.lang.OutOfMemoryError: GC overhead limit exceeded堆内存不足或单任务数据过大调小 executor 内存、加多分区、处理倾斜FetchFailedExceptionshuffle 数据拉取失败多因节点压力大增大重试次数检查磁盘和网络Task not serializable在闭包里引用了不可序列化的对象把变量提取到闭包外或实现 Serializableorg.apache.spark.SparkException: Job aborted due to stage failure通常是上游任务失败传导往上翻日志找第一个错误java.io.IOException: No space left on deviceshuffle 或日志把磁盘写满检查spark.local.dir和事件日志目录Table or view not foundmetastore 连接或库表名错误确认HADOOP_CONF_DIR和 Hive 配置7.2 排查的固定动作遇到作业失败我一般按这个顺序走效率最高。第一步看 driver 日志的最后 200 行找到第一个Exception或Error别被后面一堆级联报错带偏。Spark 的报错经常是链条式的最上面那个往往才是根因。第二步去 WebUI 看失败的 Stage 和失败的任务。点进失败任务的 stderr 日志能看到 executor 侧的真实报错。driver 日志通常只显示某个任务失败了详细原因在 executor 那边。第三步判断是数据问题还是资源问题。如果失败是偶发的重跑就过多半是资源或网络如果每次都在同一个地方失败大概率是数据或代码逻辑。第四步做最小复现。把失败的 DataFrame 取一小部分比如limit(10000)在spark-shell里跑一遍。能复现就说明是逻辑问题不能复现就是规模问题。提示在内网环境里spark-shell是排查利器。因为装不了外部调试工具很多问题只能靠 shell 里临时构造数据来验证。建议把常用的排查片段存成.scala文件放在边缘节点上需要时:load进来。还有一点日志一定要集中收集。内网环境没法用云上的日志服务我一般用 Spark 的 eventLog 加上自定义的 log4j 配置把作业日志按天写到 HDFS 上。出问题的时候直接hdfs dfs -cat捞比挨个机器查要快得多。8. 模型交付离线批量推理与结果落地8.1 模型的保存与加载训练完之后第一步是把模型持久化。MLlib 的模型支持save和load格式是 Parquet 加元数据跨版本兼容性一般所以要记录清楚训练时用的 Spark 版本。model cv_model.bestModel model.write().overwrite().save(hdfs:///models/gbt_churn_v3) # 加载 from pyspark.ml.classification import GBTClassificationModel loaded GBTClassificationModel.load(hdfs:///models/gbt_churn_v3)注意别把cv_model直接存了CrossValidatorModel会把所有折的模型都存下来体积是单个模型的 k 倍。取bestModel存通常只需要几十 MB 到几百 MB。保存路径要有版本管理。我的命名习惯是模型名_业务_版本号_日期比如gbt_churn_v3_20240601。生产环境一定要保留至少两个版本新版本上线后观察一段时间再清理旧的否则指标掉了都没法回滚。8.2 离线批量打分与结果输出批量推理在离线环境里其实很简单就是加载模型、transform、写出结果三步。# 读取待打分的特征表 score_df spark.table(dw.feature_wide).where(dt 2024-06-01) # 打分 scored model.transform(score_df) # 只需要 id 和预测结果不要把中间特征也写出去省空间 result scored.select( user_id, item_id, F.col(prediction).cast(int).alias(pred), F.round(F.col(probability)[1], 6).alias(score) ) result.write.mode(overwrite).parquet(hdfs:///output/score_20240601)probability[1]是取正类的概率这个在Vector类型里是按索引取的别写错。输出的score保留 6 位小数够了再精也没意义反而增加存储。结果的可视化在这类场景里要特别小心。内网环境里做展示常用的方案是把聚合后的统计结果导出成 CSV再用 ECharts 之类的库做本地大屏。但导出前一定要做脱敏和聚合明细级别的打分结果不能直接上大屏展示通常是按区域、按类别做计数和分布统计只展示汇总指标。8.3 模型监控与迭代节奏模型上线不是终点。离线环境里做监控基本只能靠定期的离线评估。我的做法是每周跑一次批量打分把结果和实际发生的标签做对比计算 AUC、KS 这类指标指标掉到阈值以下就触发重训。标签的获取往往滞后比如预测的是未来 30 天是否流失那至少要等 30 天才有真实标签。所以监控链路上要维护一张延迟标签表按天回填。这个工程细节比模型本身更容易被忽略但它决定了模型能不能长期可用。9. 一些踩坑后的个人体会最后分享几个我觉得比技术细节更重要的点。第一先跑通链路再谈精度。在一个全新的离线集群上做项目最大的风险不是模型效果差而是链路跑不通。我见过太多团队一上来就纠结用 GBT 还是 XGBoost结果 ETL 脚本连作业都提交不上去。正确顺序是先用十分之一的数据量跑通 Spark SQL 读表、特征工程、训练、打分的完整链路确认每个环节都通了再逐步放大数据规模。能在小数据上跑通的东西才值得在大数据上优化。第二参数别迷信教程要用自己集群的机器数去算。网上给的--executor-memory 16g --num-executors 100是在别人的机器上验证过的你的机器核数、内存、磁盘数都不一样。前面那套预留 20% 系统内存、每 executor 5 核、均分内存的算法虽然土但每次算出来都靠谱。第三数据倾斜的处理优先级永远高于算法调优。一个 Stage 从 30 分钟降到 3 分钟往往不是因为你换了模型而是因为你把热点 key 加盐了或者把一张表广播了。训练时间的大头在 I/O 和 shuffle不在算法本身。第四销掉所有不必要的collect()和show()。这两个操作会把数据拉回 driver在数据量大时直接把 driver 内存撑爆。要调试的话用limit(100).show()看结构没问题但看数据分布一定要用groupBy().count().show()这种聚合后的结果。第五内网环境里脚本和配置的版本管理比代码本身更重要。没有外网、没有现成的 CI所有的配置都是手工改的。我习惯在/opt/spark/conf下留一个README.md记录每次改了什么、为什么改、什么时候改的。半年后你回头看这份记录能救命。