利用 Spark ML 对农产品数据进行分析预测,给出的数据属性为品名,生产城市,近三天最高价格,最低价格,平均价格,产品类别,请给出python代码
时间: 2023-09-27 12:05:41 浏览: 61
下面是一个简单的Python代码示例,演示如何使用Spark ML库对农产品数据进行分析和预测。假设我们有一些关于农产品价格的数据,我们想要使用Spark ML库来构建一个线性回归模型来预测未来的价格。
```python
# 导入必要的库
from pyspark.sql import SparkSession
from pyspark.ml.regression import LinearRegression
from pyspark.ml.feature import VectorAssembler, StringIndexer, OneHotEncoder
from pyspark.ml import Pipeline
# 创建SparkSession
spark = SparkSession.builder.appName('CropPricePrediction').getOrCreate()
# 读取CSV文件并将其转换为DataFrame
df = spark.read.csv('crop_price_data.csv', header=True, inferSchema=True)
# 创建特征向量
assembler = VectorAssembler(inputCols=['highest_price', 'lowest_price', 'average_price'], outputCol='features')
df = assembler.transform(df)
# 对类别和生产城市进行编码
indexer1 = StringIndexer(inputCol='product_category', outputCol='category_indexed')
indexer2 = StringIndexer(inputCol='production_city', outputCol='city_indexed')
encoder = OneHotEncoder(inputCols=['category_indexed', 'city_indexed'], outputCols=['category_vec', 'city_vec'])
pipeline = Pipeline(stages=[indexer1, indexer2, encoder])
df = pipeline.fit(df).transform(df)
# 将数据集分成训练集和测试集
train_data, test_data = df.randomSplit([0.7, 0.3])
# 创建线性回归模型
lr = LinearRegression(featuresCol='features', labelCol='average_price')
# 训练模型
model = lr.fit(train_data)
# 在测试数据上进行预测
predictions = model.transform(test_data)
# 显示预测结果
predictions.select('productName', 'productionCity', 'highest_price', 'lowest_price', 'average_price', 'prediction').show()
# 评估模型
from pyspark.ml.evaluation import RegressionEvaluator
evaluator = RegressionEvaluator(labelCol='average_price', predictionCol='prediction', metricName='rmse')
rmse = evaluator.evaluate(predictions)
print('Root Mean Squared Error (RMSE):', rmse)
# 关闭SparkSession
spark.stop()
```
在这个示例中,我们首先读取CSV文件并将其转换为Spark DataFrame。接下来,我们使用VectorAssembler创建特征向量,并对类别和生产城市进行编码。然后,我们使用LinearRegression算法训练线性回归模型,并在测试数据上进行预测。最后,我们使用RegressionEvaluator评估模型的性能。