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

基于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_predictions
    print(f"模型集成准确率: {accuracy:.4f}")