### Pyspark 和随机森林算法预测商品销量的脚本示例 以下是基于 PySpark 的随机森林回归模型用于预测商品销量的一个完整脚本示例。此脚本涵盖了数据加载、预处理、特征工程以及模型训练和评估的过程。 #### 数据准备阶段 PySpark 支持多种数据源，例如 CSV 文件或其他分布式存储中的数据集。假设我们有一个包含历史销售记录的数据集 `sales_data.csv`，其中每一行代表一条销售记录，包括日期、产品类别、价格、促销标志以及其他可能影响销量的因素。 ```python from pyspark.sql import SparkSession from pyspark.ml.feature import VectorAssembler, StringIndexer from pyspark.ml.regression import RandomForestRegressor from pyspark.ml.evaluation import RegressionEvaluator from pyspark.ml.tuning import CrossValidator, ParamGridBuilder # 初始化 SparkSession spark = SparkSession.builder \ .appName("SalesPrediction") \ .getOrCreate() # 加载数据 data_path = "path/to/sales_data.csv" df = spark.read.format("csv").option("header", "true").load(data_path) # 查看数据结构 df.printSchema() ``` #### 缺失值处理 对于缺失值，在 Pandas 中可以通过 `.dropna()` 或者 `.fillna()` 方法来解决[^1]。而在 PySpark 中，则可以利用内置函数完成类似的逻辑： ```python # 删除含有任何 NULL 值的行 cleaned_df = df.na.drop() # 替换特定列中的 NULL 值为默认值（如均值） mean_values = cleaned_df.selectExpr( ["avg(`price`) as avg_price"] ).collect()[0].asDict() filled_df = cleaned_df.fillna(mean_values["avg_price"], subset=["price"]) ``` #### 特征转换与编码 为了使分类变量能够被机器学习模型理解并有效使用，通常需要对其进行索引化操作；而数值型输入则可以直接组合成向量形式供后续建模调用。 ```python # 对字符串类型的特征进行索引化 category_indexers = [ StringIndexer(inputCol=col, outputCol=f"{col}_index") for col in ["product_category", "promotion_flag"]] # 将所有特征组装到一个单独的特征向量中 feature_columns = [f"{c}_index" if c in {"product_category", "promotion_flag"} else c for c in filled_df.columns if c != 'sales'] vector_assembler = VectorAssembler(inputCols=feature_columns, outputCol="features") # 应用转化器创建最终的 DataFrame pipeline_stages = category_indexers + [vector_assembler] for stage in pipeline_stages: filled_df = stage.fit(filled_df).transform(filled_df) final_df = filled_df.withColumnRenamed('sales', 'label') ``` #### 构建与优化随机森林模型 通过定义参数网格搜索空间，并采用交叉验证方法寻找最佳超参配置，从而提升模型性能表现。 ```python rf = RandomForestRegressor(featuresCol="features", labelCol="label") param_grid = ParamGridBuilder()\ .addGrid(rf.numTrees, [50, 100])\ .addGrid(rf.maxDepth, [5, 10])\ .build() evaluator = RegressionEvaluator(labelCol="label", predictionCol="prediction", metricName="rmse") cross_validator = CrossValidator(estimator=rf, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=3) cv_model = cross_validator.fit(final_df) best_rf_model = cv_model.bestModel predictions = best_rf_model.transform(final_df) print("Root Mean Squared Error (RMSE): ", evaluator.evaluate(predictions)) ``` 以上即是一个完整的流程展示，从原始数据导入直至获得经过调优后的随机森林回归模型及其对应的误差评价指标 RMSE 结果输出。