PySpark集成XGBoost实战：分布式训练的依赖管理与生产部署

1. 项目概述：为什么在 PySpark 生态里硬要“塞进” XGBoost？

我干数据工程和机器学习平台支撑快十二年了，从 Hadoop MapReduce 写 Java UDF 开始，到 Spark SQL 做特征平台，再到今天用 PySpark 搭建端到端的模型训练流水线。这中间踩过最大的坑，不是算法调参失败，而是——选错了模型与执行引擎的耦合方式。XGBoost 就是典型。它不是 Spark MLlib 原生支持的算法，但业务方一句“这个模型线上 AUC 高 3 个点”，你就得把它跑通。这不是炫技，是真实产线里的生存逻辑。

XGBoost 的优势，业内早说烂了：梯度提升的工程极致、缺失值原生处理、树结构并行切分、内存友好。但这些优势，在单机 Python（scikit-learn 接口）里是“开箱即用”的；一旦数据量涨到 50GB+，单机内存扛不住，CPU 利用率卡在 100% 却只跑一个核，这时候你才真正理解什么叫“分布式训练不是加个--master yarn就完事”。PySpark 提供的是数据调度层和 DataFrame API，它不负责模型训练的数学计算——XGBoost 的核心 C++ 训练引擎，压根不认识 Spark 的 RDD 或 DataFrame。所以，所谓“PySpark + XGBoost”，本质是一场精密的“跨进程协作”：Spark 负责把数据切片、分发、聚合，XGBoost 负责在每个 Executor 上启动独立的 C++ 进程做本地训练，最后再把模型参数或预测结果传回 Driver。这不是简单的pip install xgboost就能搞定的事，它牵扯到 JVM 和 Python 进程的通信、二进制依赖的打包分发、序列化协议的兼容性，甚至 Python 版本和 Spark 版本的隐式绑定。

这篇文章要解决的，就是这个“硬塞”的全过程。它不讲 XGBoost 的数学原理（那本书够厚），也不吹嘘“一键分布式”（那是营销话术），而是聚焦在Python < 3.8 环境下，如何用最稳定、最可复现、最贴近生产环境的方式，让 XGBoost 在 PySpark 集群上真正跑起来、训出来、用得上。你会看到每一个.jar文件为什么必须放对位置，addPyFile为什么不能用--py-files替代，StringIndexer的handleInvalid="keep"背后藏着多少空值陷阱，以及 CrossValidator 在 XGBoost 场景下为何是“昂贵的奢侈品”。这不是教程，是我在三个不同客户现场，为解决贷款审批、用户流失预警、广告点击率预估这三个真实项目，反复打磨出来的“血泪操作手册”。

2. 整体架构设计与技术选型深度拆解

2.1 为什么不用 XGBoost 官方的xgboost.spark？——版本锁死的现实困境

XGBoost 从 1.7.0 版本开始，官方确实提供了xgboost.spark模块，宣称“原生支持 Spark 分布式训练”。听起来很美，对吧？但现实是，它要求 Python ≥ 3.8，且 Spark 版本需严格匹配（比如 XGBoost 1.7.6 只认 Spark 3.3.x）。我们团队去年在一个金融客户的离线平台升级时就栽在这儿：客户集群是 Spark 3.2.1 + Python 3.7.12，这是他们经过半年安全审计才批准的稳定版本。强行升级 Python？意味着所有已上线的 PySpark 作业、UDF、自定义序列化器全都要回归测试，成本远超模型收益。最终我们放弃了官方模块，回头用更底层、但兼容性更强的sparkxgb方案。这背后是一个关键认知：在企业级数据平台中，“稳定压倒一切”，而“稳定”的定义，是能与现有技术栈无缝咬合，而不是追逐最新特性。

sparkxgb是由 DMLC 社区（XGBoost 的母社区）维护的一个桥接库，它不试图改造 XGBoost 引擎，而是用 Java/Scala 编写一个 Spark Estimator，通过 JNI（Java Native Interface）调用本地 XGBoost 库。它的核心价值在于“解耦”：Spark 侧的逻辑（数据读取、特征工程、Pipeline 管理）完全用 Scala/Java 实现，保证与 Spark 运行时的 100% 兼容；而模型训练的核心计算，仍交给 XGBoost 的 C++ 引擎。这种设计，让它能向下兼容 Spark 2.4+ 和 Python 3.6+，正是我们这类“老系统焕新”场景的救命稻草。

2.2 依赖包的三重角色：JAR、ZIP、Python Wrapper 的分工逻辑

看原始资料里提到的三个文件：xgboost4j-spark.jar、xgboost4j.jar、sparkxgb.zip，很多人会疑惑：为啥要三个？能不能合并？答案是：它们各自承担着 JVM 层、Native 层、Python 层的不可替代职责，缺一不可。

• xgboost4j-spark.jar：这是 Spark 的“大脑”。它包含了XGBoostClassifier、XGBoostRegressor这些 Estimator 类，以及XGBoostClassificationModel这些 Model 类。当你在 PySpark 代码里from sparkxgb.xgboost import XGBoostClassifier时，实际加载的就是这个 JAR 包里的 Java 类。它负责解析你传入的featuresCol、labelCol参数，将 Spark DataFrame 转换成内部的数据结构，并协调整个训练流程。它必须通过--jars参数注入到 SparkContext 的 ClassPath 中，否则 Driver 根本找不到这些类。

• xgboost4j.jar：这是 XGBoost 的“肌肉”。它封装了 XGBoost 的 C++ 训练引擎的 JNI 接口。xgboost4j-spark.jar在需要真正计算时，会通过 JNI 调用这个 JAR 里的本地方法（Native Method），进而触发底层 C++ 代码。这个 JAR 里还包含了针对不同操作系统（Linux/macOS/Windows）编译的.so或.dylib动态链接库。它必须和xgboost4j-spark.jar放在同一个目录下，并一同通过--jars加载，否则 JNI 调用会因找不到本地库而崩溃。

• sparkxgb.zip：这是 Python 的“翻译官”。它不是一个普通的 Python 包，而是一个被zipimport机制加载的压缩包，里面包含了sparkxgb/xgboost.py这个 Python 模块。这个模块的作用，是提供一个 Python 风格的接口（比如XGBoostClassifier(objective="binary:logistic")），让你能在 PySpark 脚本里用熟悉的语法去配置和调用。它内部做的，是把你的 Python 参数，转换成 Java 对象，再传递给xgboost4j-spark.jar里的 Java 类。它不能用pip install安装，因为 Spark Executor 启动的是独立的 Python 进程，pip安装的包只在 Driver 进程里有效。必须用sparkContext.addPyFile()，将这个 ZIP 包分发到每一个 Executor 的 Python Path 下，确保每个 Worker 都能import sparkxgb。

提示：很多初学者会把sparkxgb.zip错误地当成一个可以pip install的包，然后在集群上pip install sparkxgb。这是大忌。pip install只影响当前 Python 解释器的 site-packages，而 Spark Executor 使用的是PYSPARK_PYTHON环境变量指定的 Python 解释器，它默认不会去读取 Driver 机器上的 pip 包。addPyFile是 Spark 提供的、专用于分发 Python 依赖的“正统”方式。

2.3 Spark Session 初始化的魔鬼细节：local[*]不等于“真分布式”

原始代码里master("local[*]")看似简单，但这是本地开发调试的“黄金配置”，也是最容易被误解的地方。local[*]表示使用本机所有 CPU 核心来模拟一个 Spark 集群。它对于验证代码逻辑、调试 Pipeline 流程、小数据集快速迭代，是无可替代的。但它的“模拟”属性，也埋下了几个深坑：

• 内存隔离失效：在真正的 YARN 或 Kubernetes 集群上，每个 Executor 进程有独立的 JVM Heap，内存溢出（OOM）只会杀死那个 Executor。而在local[*]模式下，所有“Executor”都运行在同一个 JVM 进程里，共享 Heap。这意味着，如果你的 XGBoost 训练参数（如numRound）设得过大，或者数据分片不均，一个“虚拟 Executor”的 OOM 会直接导致整个 SparkSession 崩溃，错误日志里全是java.lang.OutOfMemoryError: Java heap space，根本看不出是哪个环节的问题。

• 网络通信被绕过：local[*]模式下，Driver 和 “Executors” 之间走的是进程内通信（In-process），而非真实的 TCP/IP 网络。这会导致一些依赖于网络超时、重试机制的配置（如spark.network.timeout）完全不生效。等你把代码提交到 YARN 上，发现任务卡在Connecting to driver，才恍然大悟——原来本地没暴露这个问题。

• 依赖路径的幻觉：在local[*]下，os.environ['PYSPARK_SUBMIT_ARGS']设置的--jars路径，只要 Driver 机器上存在就行。但到了 YARN 集群，这些 JAR 文件必须能被所有 NodeManager 访问到，通常需要上传到 HDFS 或 S3，并用hdfs://或s3a://的 URI 来指定。local[*]让你忽略了这个关键的“依赖分发”步骤。

所以，我的实操心得是：永远用local[*]开发和单元测试，但每完成一个功能点，就立刻在最小化的 YARN 集群（哪怕只有 2 个 NodeManager）上跑一次端到端集成测试。不要等到所有代码写完，再一次性上集群，那会把所有问题混在一起，排查难度指数级上升。

3. 核心细节解析与实操要点：从数据加载到 Pipeline 构建

3.1 数据加载与 Schema 探查：Parquet 的双刃剑

原始示例用了spark.read.parquet('train.parquet')。Parquet 是列式存储，对 Spark 友好，读取速度快，这是优点。但它的“友好”是有前提的：Schema 必须稳定且明确。我在一个电商推荐项目里就吃过亏：上游数据管道偶尔会因为上游 ETL 逻辑 bug，写入一个字段类型为string的user_id，而正常情况应该是long。Parquet 文件本身不强制校验 Schema，Spark 读取时会按第一个文件的 Schema 作为基准，后续文件如果类型不一致，就会在show()或count()时抛出org.apache.spark.sql.AnalysisException: cannot resolve 'user_id' due to data type mismatch。这种错误非常隐蔽，因为show(5)可能只读前 5 行，恰好没碰到异常数据。

因此，我的标准操作是：在read.parquet后，立即调用printSchema()，并人工核对每一列的类型是否符合预期。对于Loan_Status这种目标变量，更要小心。原始数据是'Y'/'N'字符串，代码里用F.when(F.col('Loan_Status')=='Y',1).otherwise(0)转成1/0。这个逻辑看似正确，但如果数据里存在'y'、'yes'、'N '（带空格）等变体，==就会失效，导致otherwise(0)把所有非'Y'的值都变成0，包括null。这会让模型学到一个错误的先验：所有未知状态都等同于拒绝。正确的做法是显式处理null：

from pyspark.sql.functions import when, col, isnan, isnull # 更鲁棒的转换 data = data.withColumn( 'label', when(col('Loan_Status') == 'Y', 1) .when(col('Loan_Status') == 'N', 0) .when(isnull(col('Loan_Status')) | isnan(col('Loan_Status')), None) # 显式标记 null .otherwise(None) # 兜底，防止意外字符 )

注意：isnull()和isnan()是两个不同的函数。isnull()检查 SQLNULL，isnan()检查浮点数的NaN。字符串列理论上不会有NaN，但加上更保险。

3.2 字符串特征编码：StringIndexer的handleInvalid陷阱

原始代码里，所有StringIndexer都设置了setHandleInvalid("keep")。这是个关键选择，但它的含义常被误解。“keep” 并不是“保留原始字符串”，而是“为所有未在训练集里见过的字符串，分配一个特殊的、统一的索引值（通常是 -1.0）”。这在生产环境中至关重要。

想象一下：模型在 2023 年 1 月用历史数据训练，当时Property_Area只有'Urban','Rural','Semiurban'三个值。到了 2023 年 6 月，业务拓展到一个新的城市，数据里出现了'Metro'。如果StringIndexer的handleInvalid设为"error"（默认值），那么在对新数据做transform()时，就会直接报错java.lang.IllegalArgumentException: Field Property_Area contains invalid value: Metro，整个批处理任务失败。而设为"keep"，'Metro'就会被编码成-1.0，模型依然能做出预测（虽然可能不准，但至少不中断）。

但这里有个隐藏风险：-1.0这个值，会被VectorAssembler当作一个有效的数值特征，和其他特征一起输入 XGBoost。XGBoost 会把它当作一个“特殊类别”来学习。如果Metro出现的频率很低，这个-1.0特征的分裂增益可能很小，模型就学不到它的模式。更稳妥的做法，是在StringIndexer之后，再加一个OneHotEncoderEstimator（注意是 Estimator，不是旧版的 Encoder），它会把-1.0也编码成一个独立的二进制维度，这样模型就能明确区分“已知类别”和“未知类别”。

3.3 特征向量组装：VectorAssembler的handleInvalid与缺失值策略

VectorAssembler的handleInvalid='keep'选项，其行为与StringIndexer截然不同。它不是给缺失值分配一个特殊码，而是直接将该行的整个features向量设为null。这会导致后续的 XGBoost 训练直接失败，因为 XGBoost 的输入矩阵不能包含null。

XGBoost 本身对缺失值有强大的原生处理能力（通过missing参数指定），但它要求缺失值是以一个具体的数值（如0.0,np.nan,-999）来表示，而不是 SQL 的NULL。因此，VectorAssembler的handleInvalid必须设为"error"（默认），并在它之前，用Imputer或fillna()对所有数值特征列进行缺失值填充。

from pyspark.ml.feature import Imputer # 对数值特征进行缺失值填充 numeric_cols = ['ApplicantIncome', 'CoapplicantIncome', 'LoanAmount', 'Loan_Amount_Term', 'Credit_History'] imputer = Imputer( inputCols=numeric_cols, outputCols=numeric_cols ).setStrategy("median") # 用中位数填充，比均值对异常值更鲁棒 # Pipeline stages 顺序很重要：先填充，再索引，再组装 pipeline = Pipeline().setStages([ imputer, # 第一步：填充数值缺失值 index1, index2, index3, index4, index5, # 第二步：字符串索引 vec_assembler, # 第三步：向量组装 xgb # 第四步：模型训练 ])

实操心得：Imputer的setStrategy("median")是我在线上项目中的首选。均值（mean）容易被极端值拉偏，比如ApplicantIncome里混入一个10000000的脏数据，均值就失真了。中位数对异常值免疫，更稳健。"most_frequent"（众数）适合分类特征，但对连续数值意义不大。

4. 实操过程与核心环节实现：从零构建可复现的训练流水线

4.1 环境准备与依赖下载：版本匹配的精确计算

原始资料里只说“下载 prerequisite files”，但没说清版本对应关系。这是最容易出问题的环节。xgboost4j-spark、xgboost4j、sparkxgb三者之间，以及它们与 Spark、Scala 的版本，存在严格的兼容矩阵。我整理了一个在生产环境中验证过的、最稳定的组合（截至 2024 年中）：

下载地址（请务必使用此链接，避免 CDN 缓存旧版本）：

• xgboost4j-spark-1.0.2.jar: https://repo1.maven.org/maven2/ml/dmlc/xgboost4j-spark_2.12/1.0.2/xgboost4j-spark_2.12-1.0.2.jar
• xgboost4j-1.0.2.jar: https://repo1.maven.org/maven2/ml/dmlc/xgboost4j_2.12/1.0.2/xgboost4j_2.12-1.0.2.jar
• sparkxgb-0.9.0.zip: https://github.com/dmlc/xgboost/releases/download/v1.0.2/sparkxgb-0.9.0.zip

提示：不要从 GitHub Release 页面直接下载sparkxgb-0.9.0.zip，那个是源码包。一定要从上面的 Maven Central 链接下载，那是编译好的、可直接addPyFile的二进制包。

4.2 Spark Session 初始化：完整的、可部署的代码模板

下面是我现在所有项目里使用的、经过千锤百炼的 Spark Session 初始化模板。它不仅包含了原始代码的--jars和addPyFile，还加入了生产环境必需的健壮性配置：

import os import findspark from pyspark.sql import SparkSession from pyspark import SparkContext # 1. 设置 PYSPARK_SUBMIT_ARGS —— 这是 JVM 层的入口 # 关键：--jars 必须是逗号分隔，且路径必须是绝对路径！相对路径在集群上会失效。 jar_path = "/path/to/your/jars" # 请替换成你实际存放 JAR 的绝对路径 os.environ['PYSPARK_SUBMIT_ARGS'] = ( f'--jars {jar_path}/xgboost4j-spark_2.12-1.0.2.jar,{jar_path}/xgboost4j_2.12-1.0.2.jar ' '--driver-class-path /path/to/your/jars/xgboost4j-spark_2.12-1.0.2.jar ' '--conf spark.sql.adaptive.enabled=false ' # 关闭 AQE，XGBoost 与 AQE 兼容性不佳 'pyspark-shell' ) # 2. 初始化 findspark —— 确保能找到 Spark 安装目录 findspark.init('/path/to/your/spark') # 请替换成你 Spark 的绝对安装路径 # 3. 创建 SparkSession —— 生产环境必须显式设置 master spark = SparkSession.builder \ .appName("XGBoost-Loan-Prediction") \ .master("yarn") \ # 本地开发用 "local[*]"，生产用 "yarn" 或 "k8s://..." .config("spark.sql.adaptive.enabled", "false") \ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ # Kryo 序列化更快 .config("spark.kryoserializer.buffer.max", "512m") \ # 防止大模型序列化失败 .config("spark.sql.adaptive.coalescePartitions.enabled", "false") \ # 关闭分区合并，避免数据倾斜 .getOrCreate() # 4. 添加 Python 依赖 —— 这是 Python 层的入口 zip_path = "/path/to/your/sparkxgb-0.9.0.zip" # 请替换成你实际存放 ZIP 的绝对路径 spark.sparkContext.addPyFile(zip_path) # 5. 【可选但强烈推荐】验证依赖是否加载成功 try: from sparkxgb.xgboost import XGBoostClassifier print("✅ XGBoostClassifier 导入成功，依赖加载无误") except ImportError as e: print(f"❌ 导入失败: {e}") raise

注意：--driver-class-path这个配置经常被忽略。它告诉 Spark Driver 的 JVM，xgboost4j-spark.jar里的类应该优先从这个路径加载，避免与集群上可能存在的旧版本冲突。spark.serializer和spark.kryoserializer.buffer.max是为了应对 XGBoost 模型对象可能很大的情况（尤其是numRound很大时），防止序列化失败。

4.3 Pipeline 构建与模型训练：完整的、可复现的端到端代码

现在，把前面所有环节串起来，给出一个可以直接复制、粘贴、运行的完整脚本。这个脚本包含了所有关键注释和错误处理：

from pyspark.sql import functions as F from pyspark.sql.types import DoubleType, StringType, IntegerType from pyspark.ml import Pipeline from pyspark.ml.feature import StringIndexer, VectorAssembler, Imputer from pyspark.mllib.evaluation import MulticlassMetrics from pyspark.sql.functions import col, when, isnan, isnull # 1. 数据加载与探查 print("🔍 正在加载数据...") data = spark.read.parquet("hdfs://namenode:8020/data/loan/train.parquet") # 使用 HDFS 路径 print("📊 数据 Schema:") data.printSchema() # 2. 目标变量转换（鲁棒版） print("🔄 转换目标变量 'Loan_Status'...") data = data.withColumn( 'label', when(col('Loan_Status') == 'Y', 1) .when(col('Loan_Status') == 'N', 0) .when(isnull(col('Loan_Status')) | isnan(col('Loan_Status')), None) .otherwise(None) ) # 过滤掉 label 为 null 的行，确保训练数据干净 data = data.filter(col('label').isNotNull()) print(f"✅ 转换完成，有效样本数: {data.count()}") # 3. 特征工程 Pipeline print("⚙️ 构建特征工程 Pipeline...") # 数值特征缺失值填充 numeric_features = ['ApplicantIncome', 'CoapplicantIncome', 'LoanAmount', 'Loan_Amount_Term', 'Credit_History'] imputer = Imputer( inputCols=numeric_features, outputCols=numeric_features ).setStrategy("median") # 字符串特征索引 indexers = [ StringIndexer(inputCol="Gender", outputCol="GenderIndex", handleInvalid="keep"), StringIndexer(inputCol="Married", outputCol="MarriedIndex", handleInvalid="keep"), StringIndexer(inputCol="Education", outputCol="EducationIndex", handleInvalid="keep"), StringIndexer(inputCol="Self_Employed", outputCol="SelfEmployedIndex", handleInvalid="keep"), StringIndexer(inputCol="Property_Area", outputCol="PropertyAreaIndex", handleInvalid="keep") ] # 特征向量组装 feature_cols = ['GenderIndex', 'MarriedIndex', 'EducationIndex', 'SelfEmployedIndex', 'PropertyAreaIndex'] + numeric_features vec_assembler = VectorAssembler( inputCols=feature_cols, outputCol='features', handleInvalid='error' # 必须设为 error，确保缺失值已被填充 ) # 4. XGBoost 模型定义 print("🚀 初始化 XGBoost 模型...") from sparkxgb.xgboost import XGBoostClassifier xgb = XGBoostClassifier( objective="binary:logistic", featuresCol="features", labelCol="label", missing=0.0, # XGBoost 用 0.0 表示缺失值 numRound=100, # 训练轮数，可根据数据量调整 maxDepth=6, # 树的最大深度 eta=0.1, # 学习率 subsample=0.8, # 训练样本采样率 colsampleBytree=0.8, # 特征采样率 seed=1712 ) # 5. 构建完整 Pipeline stages = [imputer] + indexers + [vec_assembler, xgb] pipeline = Pipeline(stages=stages) # 6. 数据集划分 print("✂️ 划分训练集和测试集...") train_df, test_df = data.randomSplit([0.7, 0.3], seed=1712) print(f"✅ 训练集大小: {train_df.count()}, 测试集大小: {test_df.count()}") # 7. 模型训练 print("🔥 开始训练 XGBoost 模型...") model = pipeline.fit(train_df) print("✅ 模型训练完成！") # 8. 模型预测 print("🔮 在测试集上进行预测...") predictions = model.transform(test_df).select('Loan_ID', 'prediction', 'label') predictions.show(5) # 9. 模型评估 print("📈 计算模型评估指标...") # 将 prediction 和 label 转为 DoubleType，以适配 MulticlassMetrics pred_and_labels = predictions.select( col('prediction').cast(DoubleType()).alias('prediction'), col('label').cast(DoubleType()).alias('label') ).rdd metrics = MulticlassMetrics(pred_and_labels) cm = metrics.confusionMatrix().toArray() accuracy = (cm[0][0] + cm[1][1]) / cm.sum() precision = cm[1][1] / (cm[0][1] + cm[1][1]) if (cm[0][1] + cm[1][1]) > 0 else 0.0 recall = cm[1][1] / (cm[1][0] + cm[1][1]) if (cm[1][0] + cm[1][1]) > 0 else 0.0 print(f"🎯 Accuracy: {accuracy:.4f}") print(f"🎯 Precision: {precision:.4f}") print(f"🎯 Recall: {recall:.4f}") # 10. 【可选】保存模型 print("💾 保存训练好的 Pipeline 模型...") model.write().overwrite().save("hdfs://namenode:8020/models/xgboost_loan_v1") print("✅ 模型保存成功！")

这段代码，我已经在多个客户现场部署过。它最大的特点是可预测性：无论你在本地local[*]运行，还是提交到 YARN 集群，只要输入数据和依赖版本一致，输出的模型和指标就完全一致。这就是工程化和“玩具代码”的本质区别。

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

5.1 经典错误：“No module named 'sparkxgb'” ——addPyFile的路径玄学

这是新手遇到的第一个拦路虎。错误信息很直白，但原因五花八门。我总结了三种最高频的场景和对应的解决方案：

提示：一个快速验证addPyFile是否成功的命令是：spark.sparkContext.parallelize([1]).map(lambda x: __import__('sparkxgb')).collect()。如果返回[<module 'sparkxgb' from '...'>]，说明成功；如果报错，说明失败。

5.2 性能瓶颈：“Executor OOM” —— XGBoost 内存的双重消耗

XGBoost 训练时的内存占用，是 Spark 内存和 XGBoost 本地内存的叠加。一个 Executor 的总内存 =spark.executor.memory+XGBoost 的 native memory。后者不受 Spark 控制，很容易爆。

症状：YARN 日志里出现Container killed by YARN for exceeding memory limits，或者 Executor 日志里有java.lang.OutOfMemoryError: Direct buffer memory。

根因分析：

• spark.executor.memory只控制 JVM Heap，而 XGBoost 的 C++ 引擎使用的是 JVM 外的“直接内存”（Direct Memory）。
• xgboost4j.jar会通过ByteBuffer.allocateDirect()分配大量直接内存，用于存放训练数据矩阵和梯度直方图。

解决方案（三管齐下）：

1. 增加直接内存限额：在PYSPARK_SUBMIT_ARGS里添加-XX:MaxDirectMemorySize=4g（根据你的executor.memory按比例设置，通常是 1/2 到 1/3）。
2. 降低 XGBoost 的内存压力：减小maxDepth（深度越小，树越“瘦”，内存占用越低），增大subsample（减少每次迭代的样本量），启用tree_method='approx'（近似直方图，比exact内存友好）。
3. 增加 Executor 的总内存：--executor-memory 8g --executor-cores 4，确保有足够资源。

5.3 模型效果差：“Precision 低，Recall 高” —— 类别不平衡的无声杀手

原始示例的贷款数据，Loan_Status的分布极不均衡。假设 10000 条记录里，Y（批准）只有 2000 条，N（拒绝）有 8000 条。模型为了最大化 Accuracy，会倾向于全部预测为N，从而得到 80% 的 Accuracy，但 Precision（预测为Y的准确率）会趋近于 0。

诊断：打印混淆矩阵cm，如果cm[0][1]（把N错判为Y）很大，而cm[1][0]（把Y错判为N）也很大，说明模型在两类间摇摆，但偏向多数类。

解决方案：

• 在 XGBoost 中设置scale_pos_weight：这是一个神参数。它的值 =负样本数 / 正样本数。在我们的例子中，scale_pos_weight = 8000 / 2000 = 4.0。这告诉 XGBoost：“把一个正样本错判的代价，设为把一个负样本错判的 4 倍”。代码：XGBoostClassifier(..., scale_pos_weight=4.0)。
• 使用eval_metric='aucpr'：AUC-PR（Precision-Recall 曲线下的面积）比 AUC-ROC 更适合不平衡数据。

5.4