实验设备

PC计算机，配置Win10操作系统， VirtualBox 和 Linux系统

实验内容

1. 数据导入

从文件中导入数据，并转化为 DataFrame。
（1）解压adult.zip数据：
该数据可从林子雨spark编程基础（python版）教材官网的下载专区使用百度网盘自行下载，下面是官网网址：https://dblab.xmu.edu.cn/post/spark-python/#xiazaizhuanqu

unzip adult.zip

（2）导入数据，并转化为 DataFrame：

# 导入必要的PySpark MLlib模块
from pyspark.ml.feature import PCA  # 主成分分析(降维算法)
from pyspark.sql import Row  # 用于创建DataFrame行数据
from pyspark.ml.linalg import Vectors  # 向量操作工具(原误写为ml.inning)
from pyspark.ml.evaluation import MulticlassClassificationEvaluator  # 多分类评估器
from pyspark.ml import Pipeline, PipelineModel  # 机器学习流水线(原误写为Bigeline)

# 导入特征工程工具
from pyspark.ml.feature import (
    IndexToString,    # 将索引转换回原始字符串
    StringIndexer,    # 字符串特征索引化
    VectorIndexer,    # 向量特征索引化
    HashingTF,        # 特征哈希化
    Tokenizer         # 文本分词器
)

# 导入分类算法和模型
from pyspark.ml.classification import (
    LogisticRegression,              # 逻辑回归
    LogisticRegressionModel,         # 逻辑回归模型
    BinaryLogisticRegressionSummary  # 二分类结果摘要
)

# 导入SQL函数和模型调优工具
from pyspark.sql import functions as F  # Spark SQL函数(原误写为Functions)
from pyspark.ml.tuning import (         # 模型调优工具(原误写为ml.running)
    CrossValidator,    # 交叉验证
    ParamGridBuilder   # 参数网格构建器
)

def f(x):
    """
    数据转换函数：将原始数据行转换为包含特征向量和标签的字典
    
    参数:
        x: 输入数据行(列表/数组形式)
    
    返回:
        dict: 包含两个键的字典
            - 'features': 由指定列组成的密集向量(Vector)
            - 'label': 从x[14]提取的字符串标签
    """
    rel = {}
    # 创建特征向量：从输入行选择特定列(0,2,4,10,11,12)并转换为浮点数
    rel['features'] = Vectors.dense([
        float(x[0]),   # 第0列 → 特征1
        float(x[2]),   # 第2列 → 特征2
        float(x[4]),   # 第4列 → 特征3
        float(x[10]),  # 第10列 → 特征4
        float(x[11]),  # 第11列 → 特征5
        float(x[12])   # 第12列 → 特征6
    ])
    # 设置标签：将第14列作为分类标签
    rel['label'] = str(x[14])
    return rel

# 读取训练数据
df = spark.sparkContext.textFile("file:///usr/local/spark/mllib/adult.data") \
    .map(lambda p: Row(**f(p))) \
    .toDF()

# 读取测试数据
test = spark.sparkContext.textFile("file:///usr/local/spark/mllib/adult.test") \
    .map(lambda p: Row(**f(p))) \
    .toDF()

2. 进行主成分分析（PCA）

对 6 个连续型的数值型变量进行主成分分析。PCA（主成分分析）是通过正交变换把一组相关变量的观测值转化成一组线性无关的变量值，即主成分的一种方法。PCA 通过使用主成分把特征向量投影到低维空间，实现对特征向量的降维。请通过 setK()方法将主成分数量设置为 3，把连续型的特征向量转化成一个 3 维的主成分。

# 初始化PCA模型，设置参数：
# - k=3: 保留3个主成分
# - inputCol="features": 指定输入特征列名
# - outputCol="pcaFeatures": 指定输出列名
pca = PCA(k=3, inputCol="features", outputCol="pcaFeatures").fit(df)  # 在训练数据df上拟合PCA模型

# 应用PCA转换（注意：此处会显示BLAS/LAPACK的警告，属于正常现象，不影响计算）
result = pca.transform(df)       # 对训练数据进行降维转换
testdata = pca.transform(test)  # 对测试数据进行降维转换

# 展示转换结果（显示完整内容不截断）
result.show(truncate=False)

# 展示PCA转换后的测试集结果（完整显示不截断）
testdata.show(truncate=False)

3. 训练分类模型并预测居民收入

在主成分分析的基础上，采用逻辑斯蒂回归，或者决策树模型预测居民收入是否超过50K；对 Test 数据集进行验证。

# 1. 标签索引化：将字符串标签转换为数值索引
labelIndexer = StringIndexer(
    inputCol="label",      # 原始标签列名（包含字符串如"<=50K", ">50K"）
    outputCol="indexedLabel"  # 输出数值化标签列名
).fit(result)  # 在result DataFrame上拟合标签编码器

# 查看建立的标签映射关系（类别→索引）
for label in labelIndexer.labels:  # 注意：原代码拼写错误为label:sprint(label)
    print(label)
# 输出示例：
# <=50K  (可能对应索引0)
# >50K   (可能对应索引1)

# 2. 特征向量索引化：对PCA降维后的特征进行处理
featureIndexer = VectorIndexer(
    inputCol="pcaFeatures",     # 输入特征列（PCA降维后的3维向量）
    outputCol="indexedFeatures", # 输出处理后的特征列
    maxCategories=4             # 可选：设置哪些特征视为类别型（此处未显式设置）
).fit(result)  # 在result DataFrame上拟合特征索引器

# 查看处理后的特征维度（应与PCA设置的k值一致）
print(featureIndexer.numFeatures)  # 输出: 3

# 1. 准备标签转换器（将预测的数值索引转回原始字符串标签）
labelConverter = IndexToString(
    inputCol="prediction",       # 模型预测的数值标签列
    outputCol="predictedLabel",  # 转换后的字符串标签列
    labels=labelIndexer.labels   # 使用之前StringIndexer的标签映射
)

# 2. 配置逻辑回归模型
lr = LogisticRegression() \
    .setLabelCol("indexedLabel")     # 设置数值化标签列 \
    .setFeaturesCol("indexedFeatures")  # 设置处理后的特征列

# 3. 构建完整机器学习流水线
lrPipeline = Pipeline() \
    .setStages([
        labelIndexer,    # 第一步：标签索引化
        featureIndexer,  # 第二步：特征索引化
        lr,             # 第三步：逻辑回归模型
        labelConverter   # 第四步：标签反向转换
    ])

# 4. 训练模型
lrPipelineModel = lrPipeline.fit(result)  # 在训练数据上拟合

# 5. 提取逻辑回归模型组件并打印参数
lrModel = lrPipelineModel.stages[2]  # 获取流水线第三阶段（逻辑回归模型）
print("Coefficients: \n" + str(lrModel.coefficientMatrix) +
      "\nIntercept: " + str(lrModel.interceptVector) +
      "\nnumClasses: " + str(lrModel.numClasses) +
      "\nnumFeatures: " + str(lrModel.numFeatures))

# 输出结果：
# Coefficients: 
#   DenseMatrix([[-1.98285864e-07, -3.50909247e-04, -8.45150628e-04]])
# Intercept: [-1.4525992557843347]
# numClasses: 2
# numFeatures: 3

# 6. 在测试集上预测
lrPredictions = lrPipelineModel.transform(testdata)

# 7. 模型评估
evaluator = MulticlassClassificationEvaluator() \
    .setLabelCol("indexedLabel") \     # 真实标签列
    .setPredictionCol("prediction")    # 预测标签列

lrAccuracy = evaluator.evaluate(lrPredictions)  # 计算准确率
print("Test Error = %g " % (1.0 - lrAccuracy))  # 输出测试错误率
# Test Error = 0.223576 

4. 超参数调优

利用 CrossValidator 确定最优的参数，包括最优主成分 PCA 的维数、分类器自身的参数等。

from pyspark.ml.feature import PCA
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

# 1. 初始化PCA（主成分分析）
pca = PCA() \
    .setInputCol("features") \      # 输入特征列名
    .setOutputCol("pcaFeatures")   # 输出降维后特征列名

# 2. 标签索引化（字符串标签→数值索引）
labelIndexer = StringIndexer() \
    .setInputCol("label") \        # 原始标签列
    .setOutputCol("indexedLabel")  # 数值化标签列
    .fit(df)  # 注意：应先fit数据

# 3. 特征索引化（自动识别类别型特征）
featureIndexer = VectorIndexer() \
    .setInputCol("pcaFeatures") \   # PCA降维后的特征
    .setOutputCol("indexedFeatures") \  # 修正拼写错误：原为indexedFeatured
    .setMaxCategories(4)  # 特征值≤4时视为类别型

# 4. 标签反向转换器（预测值→原始标签）
labelConverter = IndexToString() \
    .setInputCol("prediction") \    # 模型预测的数值
    .setOutputCol("predictedLabel") \  # 转换后的字符串标签
    .setLabels(labelIndexer.labels)  # 使用之前的标签映射

# 5. 逻辑回归模型配置
lr = LogisticRegression() \
    .setLabelCol("indexedLabel") \  # 修正拼写：原为featureeCol
    .setFeaturesCol("indexedFeatures")

# 6. 构建完整流水线
lrPipeline = Pipeline() \
    .setStages([
        pca,              # 第一步：PCA降维
        labelIndexer,     # 第二步：标签索引化
        featureIndexer,   # 第三步：特征索引化
        lr,              # 第四步：逻辑回归
        labelConverter    # 第五步：标签反向转换
    ])  # 修正拼写：原包含converse

# 7. 参数网格配置
paramGrid = ParamGridBuilder() \
    .addGrid(pca.k, [1, 2, 3, 4, 5, 6]) \  # PCA降维维度候选值
    .addGrid(lr.elasticNetParam, [0.01, 0.1, 0.5]) \  # 弹性网络参数
    .addGrid(lr.regParam, [0.01, 0.1, 0.5]) \  # 正则化强度
    .build()  # 修正拼写：原为eLastioNetParamMap等

# 8. 交叉验证配置
cv = CrossValidator() \
    .setEstimator(lrPipeline) \  # 设置流水线
    .setEvaluator(
        MulticlassClassificationEvaluator()  # 修正拼写：原为Evolution
            .setLabelCol("indexedLabel") \
            .setPredictionCol("prediction")
    ) \
    .setEstimatorParamMaps(paramGrid) \  # 参数网格
    .setNumFolds(3)  # 3折交叉验证

# 9. 运行交叉验证
cvModel = cv.fit(df)

# 10. 在测试集上进行预测
lrPredictions = cvModel.transform(test)

# 11. 模型评估
evaluator = MulticlassClassificationEvaluator() \
    .setLabelCol("indexedLabel") \
    .setPredictionCol("prediction")  # 修正：原代码setPredicted不完整

lrAccuracy = evaluator.evaluate(lrPredictions)
print("准确率为" + str(lrAccuracy))  # 输出示例：准确率为0.7826659807123828

# 12. 获取最佳模型并分析参数
bestModel = cvModel.bestModel
lrModel = bestModel.stages[3]  # 获取流水线中的逻辑回归模型

# 打印模型参数
print("Coefficients: \n" + str(lrModel.coefficientMatrix) +
      "\nIntercept: " + str(lrModel.interceptVector) +
      "\nnumClasses: " + str(lrModel.numClasses) +
      "\nnumFeatures: " + str(lrModel.numFeatures))

# 输出示例：
# Coefficients:
#   DenseMatrix([[ 0.00000000e+00, -1.79580143e-04, -8.27746408e-04,
#     4.71538341e-02, 2.82311058e-02, -2.72876939e-01]])
# Intercept: [-7.208325318864264]
# numClasses: 2
# numFeatures: 6

# 13. 获取最佳PCA参数（可选）
bestPCA = bestModel.stages[0]
print("最佳PCA降维维度:", bestPCA.getK())

# 获取最佳PCA模型
pcaModel = bestModel.stages[0] 
# 打印主成分矩阵（6x6，因原始特征维度为6）
print("Principal Components Matrix:\n" + str(pcaModel.pc))
# 输出：
# DenseMatrix(6x6) 每列代表一个主成分，每行对应原始特征