武汉网站程序外包百度ai营销中国行
基于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}")
五、结果与总结
随机森林模型比线性回归模型表现更好的原因可以归结为以下几点：
1、随机森林能捕捉复杂的非线性关系和特征交互，而线性回归仅限于线性关系。
2、随机森林对噪声和异常值更加鲁棒，线性回归受离群值影响较大。
3、随机森林通过集成多棵决策树，降低了偏差和方差，同时提升了预测的准确性和鲁棒性。
因此，在像气象预测这样具有复杂非线性关系的数据集上，随机森林是更合适的选择。
六、完整代码与实现
完整代码可见本文顶部，可根据需求调整相关参数，并扩展到更复杂的天气预测任务中。通过 Spark 的强大并行计算能力，我们可以快速处理海量数据并训练复杂模型，为生产环境提供支持。