1. 项目背景与数据集介绍Adult数据集又称人口普查收入数据集是机器学习领域的经典数据集常用于二分类任务。这个数据集包含1994年美国人口普查数据目标是通过14个人口统计学特征预测居民年收入是否超过5万美元。数据集包含32,561条记录其中约24%的样本标记为高收入50K。数据集中的特征可分为两类连续型特征年龄、教育年限、资本收益、资本损失、每周工作时长等类别型特征工作类型、教育程度、婚姻状况、职业、种族等在实际业务中这类收入预测模型可应用于金融机构的信用风险评估精准营销的目标客户筛选社会经济学研究中的群体分析2. 环境准备与数据加载2.1 基础环境配置推荐使用以下环境配置Spark版本3.4.0MLlib的Pipeline API在此版本后更稳定Java版本JDK 1.8或11资源分配建议至少4核CPU和8GB内存# Spark启动示例本地模式 ./bin/spark-shell \ --master local[4] \ --driver-memory 4G \ --executor-memory 4G2.2 数据加载与预处理原始数据需要处理两个常见问题测试集标签带有额外句点如50K.缺失值处理原始数据中用?表示import org.apache.spark.sql.functions._ import org.apache.spark.ml.linalg.Vectors // 定义数据结构 case class Adult( age: Double, workclass: String, fnlwgt: Double, education: String, education_num: Double, marital_status: String, occupation: String, relationship: String, race: String, sex: String, capital_gain: Double, capital_loss: Double, hours_per_week: Double, native_country: String, income: String ) // 数据加载 val rawData spark.read .option(header, false) .option(inferSchema, true) .csv(adult.data.txt) .toDF(Adult.schema.fieldNames: _*) // 处理缺失值 val cleanData rawData.na.fill(Map( workclass - Unknown, occupation - Unknown, native_country - Unknown )) // 测试集标签处理 val testData spark.read.textFile(adult.test.txt) .map(_.replaceAll(50K., 50K).replaceAll(50K., 50K)) // 后续转换逻辑与训练集相同3. 特征工程实战3.1 特征选择与转换选择6个关键连续型特征进行PCA降维年龄age教育年限education_num资本收益capital_gain资本损失capital_loss每周工作时长hours_per_week最终权重fnlwgtimport org.apache.spark.ml.feature.{VectorAssembler, StandardScaler, PCA} // 特征向量组装 val assembler new VectorAssembler() .setInputCols(Array(age, education_num, capital_gain, capital_loss, hours_per_week, fnlwgt)) .setOutputCol(rawFeatures) // 特征标准化PCA前必需步骤 val scaler new StandardScaler() .setInputCol(rawFeatures) .setOutputCol(scaledFeatures) .setWithStd(true) .setWithMean(true) // PCA降维 val pca new PCA() .setInputCol(scaledFeatures) .setOutputCol(pcaFeatures) .setK(3) // 降维到3维3.2 类别特征处理对7个类别型特征采用StringIndexer OneHotEncoder组合import org.apache.spark.ml.feature.{StringIndexer, OneHotEncoder} val categoricalCols Array(workclass, education, marital_status, occupation, relationship, race, sex) // 类别转索引 val indexers categoricalCols.map { col new StringIndexer() .setInputCol(col) .setOutputCol(s${col}_index) .setHandleInvalid(keep) } // 独热编码 val encoders categoricalCols.map { col new OneHotEncoder() .setInputCol(s${col}_index) .setOutputCol(s${col}_vec) }4. 模型训练与评估4.1 逻辑回归模型构建包含特征处理和模型的完整Pipelineimport org.apache.spark.ml.classification.LogisticRegression import org.apache.spark.ml.Pipeline // 标签索引化 val labelIndexer new StringIndexer() .setInputCol(income) .setOutputCol(label) // 逻辑回归配置 val lr new LogisticRegression() .setFeaturesCol(pcaFeatures) .setLabelCol(label) .setMaxIter(100) .setRegParam(0.01) // 构建Pipeline val pipeline new Pipeline() .setStages( Array(assembler, scaler, pca) indexers encoders Array(labelIndexer, lr) ) // 训练模型 val model pipeline.fit(cleanData)4.2 决策树模型作为对比我们同时实现决策树模型import org.apache.spark.ml.classification.DecisionTreeClassifier val dt new DecisionTreeClassifier() .setFeaturesCol(pcaFeatures) .setLabelCol(label) .setMaxDepth(5) .setImpurity(gini) val dtPipeline new Pipeline() .setStages( Array(assembler, scaler, pca) indexers encoders Array(labelIndexer, dt) ) val dtModel dtPipeline.fit(cleanData)4.3 模型评估使用多分类评估器计算准确率import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator // 在测试集上预测 val lrPredictions model.transform(testData) val dtPredictions dtModel.transform(testData) // 评估器配置 val evaluator new MulticlassClassificationEvaluator() .setLabelCol(label) .setPredictionCol(prediction) .setMetricName(accuracy) // 计算准确率 val lrAccuracy evaluator.evaluate(lrPredictions) val dtAccuracy evaluator.evaluate(dtPredictions) println(f逻辑回归准确率: ${lrAccuracy}%.4f) println(f决策树准确率: ${dtAccuracy}%.4f)典型输出结果逻辑回归准确率: 0.7833 决策树准确率: 0.81245. 超参数调优5.1 交叉验证流程使用CrossValidator进行网格搜索import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator} // 参数网格 val paramGrid new ParamGridBuilder() .addGrid(pca.k, Array(3, 4, 5, 6)) // 测试不同PCA维度 .addGrid(lr.regParam, Array(0.01, 0.1, 0.5)) // 正则化参数 .addGrid(lr.elasticNetParam, Array(0.2, 0.8)) // L1/L2混合比例 .build() // 交叉验证配置 val cv new CrossValidator() .setEstimator(pipeline) .setEvaluator(evaluator) .setEstimatorParamMaps(paramGrid) .setNumFolds(3) // 3折交叉验证 .setParallelism(4) // 并行任务数 // 执行调优 val cvModel cv.fit(cleanData)5.2 最优模型分析提取最佳模型参数val bestModel cvModel.bestModel.asInstanceOf[PipelineModel] val bestPCA bestModel.stages(2).asInstanceOf[PCAModel] val bestLR bestModel.stages.last.asInstanceOf[LogisticRegressionModel] println(s最优PCA维度: ${bestPCA.getK}) println(s最优正则化参数: ${bestLR.getRegParam}) println(s最优ElasticNet参数: ${bestLR.getElasticNetParam})5.3 特征重要性分析对于决策树模型可以分析特征重要性val featureImportance dtModel.stages.last .asInstanceOf[DecisionTreeClassificationModel] .featureImportances assembler.getInputCols.zip(featureImportance.toArray) .sortBy(-_._2) .foreach { case (feat, imp) println(f$feat 重要性: ${imp}%.4f) }6. 生产化建议6.1 模型持久化将训练好的模型保存以便后续使用// 保存最佳模型 cvModel.write.overwrite().save(hdfs://path/to/bestModel) // 加载模型 val loadedModel CrossValidatorModel.load(hdfs://path/to/bestModel)6.2 性能优化技巧数据缓存对频繁使用的DataFrame调用.cache()并行度调整通过spark.sql.shuffle.partitions控制shuffle分区数广播小表对小型维度表使用广播变量// 广播小表示例 val smallTable spark.table(dim_table) spark.sparkContext.broadcast(smallTable)6.3 监控与迭代建议建立以下监控机制定期用新数据验证模型性能衰减设置特征漂移检测如PSI指标记录每次实验的超参数和评估结果// 记录实验元数据示例 val experimentLog Map( timestamp - System.currentTimeMillis(), modelType - LogisticRegression, pcaDimensions - bestPCA.getK, accuracy - evaluator.evaluate(cvModel.transform(testData)) )
)
