from pyspark.ml.feature import PCA, VectorAssembler from pyspark.ml.classification import LinearSVC from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.sql import Row , SparkSession from pyspark import SparkConf # 1.创建spark对象 spark = SparkSession.builder.config(conf = SparkConf()).getOrCreate() # fnlwgt : final-weight 样本权重 # 2.读取数据集 dataPath = "file:///home/adult.data" data = spark.read.format("csv").option("header", "true").load(dataPath) # continuous_vars = ["age","fnlwgt","education-num","capital-gain","capital-loss","hours-per-week"] # 3.数据集预处理(将六个连续型变量提取出来并转化为特征向量) assembler = VectorAssembler(inputCols=["age", "fnlwgt", "education-num", "capital-gain", "capital-loss", "hours-per-week"], outputCol="features") data = assembler.transform(data) # 4.主成分分析 pca = PCA(k=3, inputCol="features", outputCol="pca_features") model = pca.fit(data) data = model.transform(data) # 5.划分训练集和测试集 train_data, test_data = data.randomSplit([0.8, 0.2], seed=123) # 6.构建 svm 模型 svm = LinearSVC(labelCol="label", featuresCol="pca_features") # 7.参数调优 evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label", metricName="areaUnderROC") paramGrid = ParamGridBuilder().addGrid(svm.regParam, [0.1, 0.01]).addGrid(svm.maxIter, [10, 100]).addGrid(pca.k, [2, 3]).build() cv = CrossValidator(estimator=svm, evaluator=evaluator, estimatorParamMaps=paramGrid, numFolds=3) cv_model = cv.fit(train_data)

`ModuleNotFoundError: No module named 'imblearn'` 错误表明你的Python环境中没有安装 `imbalanced-learn` 库。`imbalanced-learn` 是一个用于处理类别不平衡问题的第三方库，提供了诸如SMOTE（Synthetic Minority Over-sampling Technique）等算法。 以下是解决此问题的方法： ### 解决方法 1. **安装 `imbalanced-learn` 库**： 使用 `pip` 或 `conda` 安装该库即可。 ```python # 使用pip安装 !pip install imbalanced-learn # 或者使用conda安装 !conda install -c conda-forge imbalanced-learn ``` 2. **检查安装是否成功**： 安装完成后，可以通过以下代码检查是否成功导入： ```python import imblearn print(f"imbalanced-learn version: {imblearn.__version__}") ``` 3. **确保环境一致**： 如果你使用的是Jupyter Notebook或类似环境，请确保安装的库与运行环境一致。例如，如果你在虚拟环境中运行代码，请激活对应的虚拟环境后再安装库。 --- ### 改进后的代码示例 以下是改进后的完整代码，确保 `imbalanced-learn` 已正确安装： ```python from pyspark.ml.feature import Imputer from pyspark.ml.classification import GBTClassifier from pyspark.ml.tuning import TrainValidationSplit from imblearn.over_sampling import SMOTE # 需要安装imbalanced-learn库 # 数据预处理 - 增加缺失值填充 imputer = Imputer(inputCols=numeric_features, outputCols=[f"{c}_imputed" for c in numeric_features]) # 特征组装 - 包含缺失值填充后的数值特征 assembler = VectorAssembler( inputCols=[c+"_enc" for c in categorical_features] + [f"{c}_imputed" for c in numeric_features], outputCol="raw_features" ) # PCA降维 - 调整k值为5 pca = PCA(k=5, inputCol="scaled_features", outputCol="pca_features") # 梯度提升树分类器 gbt = GBTClassifier(featuresCol="scaled_features", labelCol="label") # 参数网格 - 更精细的参数搜索 paramGrid = (ParamGridBuilder() .addGrid(gbt.maxIter, [10, 20]) # 最大迭代次数 .addGrid(gbt.maxDepth, [5, 7]) # 树深度 .addGrid(gbt.stepSize, [0.1, 0.05]) # 学习率 .build()) # 训练验证分离 train_val_split = TrainValidationSplit(estimator=gbt, estimatorParamMaps=paramGrid, evaluator=evaluator_auc, trainRatio=0.8) # 80%训练集 # 构建完整Pipeline preprocessing_pipeline = Pipeline(stages=string_indexers + one_hot_encoders + [imputer, assembler, scaler, pca]) # 数据分割与转换 train, test = data.randomSplit([0.7, 0.3], seed=42) preprocessor = preprocessing_pipeline.fit(train) train_processed = preprocessor.transform(train) test_processed = preprocessor.transform(test) # 处理类别不平衡 - 使用SMOTE smote = SMOTE(random_state=42) train_pd = train_processed.toPandas() # 转换为Pandas DataFrame X_resampled, y_resampled = smote.fit_resample(train_pd.drop("label", axis=1), train_pd["label"]) train_resampled = spark.createDataFrame(X_resampled.join(y_resampled)) # 转回Spark DataFrame # 模型训练与评估 print("=== 梯度提升树 ===") cv_model = train_val_split.fit(train_resampled) cv_predictions = cv_model.transform(test_processed) print(f"最佳模型AUC: {evaluator_auc.evaluate(cv_predictions):.4f}") print(f"准确率: {evaluator_acc.evaluate(cv_predictions):.4f}") spark.stop() ``` --- ### 解释 1. **错误原因**：`imbalanced-learn` 库未安装，导致无法导入 `SMOTE` 模块。 2. **解决方法**：通过 `pip` 或 `conda` 安装 `imbalanced-learn` 库，并确保安装成功。 3. **代码改进**：在数据预处理阶段增加了 `Imputer` 和 `SMOTE`，以处理缺失值和类别不平衡问题。 ---

首先需要下载UCI的adult数据集并安装pyspark库，然后可以按照以下步骤进行操作： 1. 导入必要的库 ```python from pyspark.ml.feature import PCA, VectorAssembler from pyspark.ml.classification import LinearSVC from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator ``` 2. 读取数据集 ```python data = spark.read.format("csv").option("header", "true").load("path/to/adult.csv") ``` 3. 数据预处理 将六个连续型变量提取出来并转化为特征向量： ```python assembler = VectorAssembler(inputCols=["age", "fnlwgt", "education_num", "capital_gain", "capital_loss", "hours_per_week"], outputCol="features") data = assembler.transform(data) ``` 4. 主成分分析 ```python pca = PCA(k=3, inputCol="features", outputCol="pca_features") model = pca.fit(data) data = model.transform(data) ``` 5. 划分训练集和测试集 ```python train_data, test_data = data.randomSplit([0.8, 0.2], seed=123) ``` 6. 构建SVM模型 ```python svm = LinearSVC(labelCol="label", featuresCol="pca_features") ``` 7. 参数调优 ```python evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label", metricName="areaUnderROC") paramGrid = ParamGridBuilder().addGrid(svm.regParam, [0.1, 0.01]).addGrid(svm.maxIter, [10, 100]).addGrid(pca.k, [2, 3]).build() cv = CrossValidator(estimator=svm, evaluator=evaluator, estimatorParamMaps=paramGrid, numFolds=3) cv_model = cv.fit(train_data) ``` 8. 预测 ```python predictions = cv_model.transform(test_data) ``` 9. 评估 ```python print("Area under ROC curve: ", evaluator.evaluate(predictions)) ```