当前位置：首页 > news >正文

基于PySpark 使用线性回归、随机森林以及模型融合实现天气预测

news 2025/8/18 6:46:49

基于PySpark 实现天气预测与模型集成

在大数据分析与机器学习领域，Spark 提供了强大的计算能力和灵活的扩展性。本文将介绍如何利用 PySpark 完成以下任务：

1、数据预处理：清洗和编码天气数据。
2、特征工程：合并数值和分类特征。
3、模型训练与评估：构建线性回归和随机森林模型。
4、模型集成：通过投票机制提升预测准确性。
以下是完整的代码和每一步的实现细节。

一、数据预处理

读取与清洗数据
我们使用 weatherAUS.csv 数据集，其中包含与澳大利亚各地天气相关的特征，例如温度、降雨量、湿度等。预处理包括：

将缺失值替换为 None。
删除含有缺失值的行。

def get_prepared_data():# 创建Spark会话spark = SparkSession.builder \.appName("WeatherPrediction") \.master("local[*]") \.getOrCreate()# 读取CSV数据df = spark.read.csv("weatherAUS.csv", header=True, inferSchema=True)df = df.select([when(col(c) == 'NA', None).otherwise(col(c)).alias(c) for c in df.columns])df = df.dropna()# 打印数据摘要numeric_cols = [...]  # 数值列列表df.describe(numeric_cols).show()return df

在这里插入图片描述

编码分类变量
分类特征通过 StringIndexer 和 OneHotEncoder 转换为数值表示，用于后续机器学习模型的训练。

使用 StringIndexer 进行编码

categorical_cols = ['Location', 'WindGustDir', 'WindDir9am', 'WindDir3pm', 'RainToday']
indexers = [StringIndexer(inputCol=col_name, outputCol=col_name + "_index") for col_name in categorical_cols]

使用 OneHotEncoder 转换为独热编码

encoders = [OneHotEncoder(inputCol=col_name + "_index", outputCol=col_name + "_onehot") for col_name in categorical_cols]

在这里插入图片描述

二、特征工程

所有数值和独热编码特征通过 VectorAssembler 合并为一个特征向量列 features，以供模型使用。

feature_cols = numeric_cols + [col_name + "_onehot" for col_name in categorical_cols]
assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
df = assembler.transform(df)

在这里插入图片描述

三、模型训练与评估

线性回归模型
线性回归用于预测天气，目标是 RainTomorrow 是否下雨（0或1）。我们使用 RegressionEvaluator 计算模型的均方根误差 (RMSE)。

def liner_model_train(df):lr = LinearRegression(labelCol="RainTomorrow_index", featuresCol="features")train_df, test_df = df.randomSplit([0.8, 0.2], seed=1234)lr_model = lr.fit(train_df)predictions = lr_model.transform(test_df)rmse = RegressionEvaluator(labelCol="RainTomorrow_index", metricName="rmse").evaluate(predictions)print(f"线性回归 RMSE: {rmse:.4f}")return predictions

在这里插入图片描述

随机森林模型
随机森林是一种强大的分类模型，可处理复杂的非线性关系。我们使用 BinaryClassificationEvaluator 评估其准确率。

def random_forest_train(df):rf = RandomForestClassifier(labelCol="RainTomorrow_index", featuresCol="features", numTrees=100)train_df, test_df = df.randomSplit([0.8, 0.2], seed=1234)model = rf.fit(train_df)predictions = model.transform(test_df)accuracy = BinaryClassificationEvaluator(labelCol="RainTomorrow_index").evaluate(predictions)print(f"随机森林 准确率: {accuracy:.4f}")return predictions

在这里插入图片描述

四、模型集成与投票

通过结合线性回归和随机森林的预测结果，我们使用投票机制提升模型的总体表现。

def model_voting(df):rf_predictions = random_forest_train(df)lr_predictions = liner_model_train(df)# 投票机制df_predictions = lr_predictions.alias("lr").join(rf_predictions.alias("rf"), on="id", how="inner") \.withColumn("final_prediction",when((col("lr_prediction") >= 0.5) & (col("rf_prediction") == 1), 1).when((col("lr_prediction") < 0.5) & (col("rf_prediction") == 0), 0).otherwise(col("rf_prediction")))# 计算准确率correct_predictions = df_predictions.filter(col("final_prediction") == col("RainTomorrow_index")).count()total_predictions = df_predictions.count()accuracy = correct_predictions / total_predictionsprint(f"模型集成准确率: {accuracy:.4f}")