背景：抽取hive上的数据，搭建线性回归模型，进行预测。
目标：抽取hive数据，并进行预测。
一、数据抽取
本次为测试数据，是通过Yarn配置的spark环境。抽取方式与spark集群抽取方式一致。也可以hdfs上准备txt文件进行读取。
import os
import sys
# 如果当前代码文件运行测试需要加入修改路径，避免出现后导包问题
BASE_DIR = os.path.dirname(os.path.dirname(os.getcwd()))
sys.path.insert(0, os.path.join(BASE_DIR))
PYSPARK_PYTHON = "/opt/anaconda3/envs/pythonOnYarn/bin/python3.8"
# 当存在多个版本时，不指定很可能会导致出错
os.environ["PYSPARK_PYTHON"] = PYSPARK_PYTHON
os.environ["PYSPARK_DRIVER_PYTHON"] = PYSPARK_PYTHON
os.environ["HADOOP_USER_NAME"]="hdfs"
from pythonOnYarn.test import SparkSessionBase
class TestPythonOnYarn(SparkSessionBase):
SPARK_APP_NAME = "testPythonOnYarn"
SPARK_URL = "yarn"
ENABLE_HIVE_SUPPORT = True
def __init__(self):
self.spark = self._create_spark_session()
oa = TestPythonOnYarn()
df = oa.spark.sql("select * from test.u_data limit 10")
读取df 数据样例如下
二、非向量数据格式问题
尝试与pandas中dataframe一样的处理方式，先拼接所需要的自变量。
from pyspark.sql.functions import split, explode, concat, concat_ws
df_ws = df.withColumn("s", concat_ws( ',',df['userid'],df['movieid'], df['rating'] ))
df_ws = df_ws.withColumn("features", split(df_ws['s'], ","))
vhouse_df = df_ws.select(['unixtime', 'features'])
vhouse_df.show(3)
处理后数据如下
直接使用该数据，代入 pyspark.ml.regression中的LinearRegression模型
from pyspark.ml.regression import LinearRegression
from pyspark.sql.session import SparkSession
lin_reg = LinearRegression(featuresCol='features',labelCol='userid',predictionCol='prediction')
lin_reg_model = lin_reg.fit(train_df)
直接运行会报错：IllegalArgumentException: requirement failed: Column s2 must be of type class org.apache.spark.ml.linalg.VectorUDT:struct<type:tinyint,size:int,indices:array<int>,values:array<double>> but was actually class org.apache.spark.sql.types.ArrayType:array<string>.
三、数据格式处理为向量格式
经过查询发现以上报错是因为列 ‘features’ 必须为向量格式才能代入模型中使用。那么尝试把特征存储在Vector中，创建一个Iris模式的RDD，然后转化成dataframe。
from pyspark.sql import Row,functions
from pyspark.ml.linalg import Vector,Vectors
#将特征列转换成向量格式，并将数据处理为浮点数形式
def f(x):
rel = {}
rel['features'] = Vectors.dense(float(x[0]),float(x[1]),float(x[2]))
rel['label'] = float(x[3])
return rel
df1 = df.rdd.map(lambda p: Row(**f(p))).toDF()
df1.show()
处理后数据样式
df1
或者使用，效果一致。
from pyspark.ml.linalg import Vector
from pyspark.ml.feature import VectorAssembler
vec = VectorAssembler(inputCols=['userid', 'movieid', 'rating'], outputCol='features')
features_df = vec.transform(df)
features_df.show()
df2
再次代入模型中进行计算，能正常运行了。
#切分数据集，70%数据作为训练集，30%数据作为测试集。
splits = df1.randomSplit([0.7, 0.3])
train_df = splits[0]
test_df = splits[1]
from pyspark.ml.regression import LinearRegression
from pyspark.sql.session import SparkSession
lin_reg=LinearRegression(featuresCol='features',labelCol='label',predictionCol='prediction')
lin_reg_model = lin_reg.fit(train_df)
四、构建和训练线性回归模型
评估线性回归模型在训练数据上的性能。
from pyspark.ml.regression import LinearRegression
lin_reg = LinearRegression(labelCol='label')
lr_model = lin_reg.fit(train_df)
lr_model.coefficients
lr_model.intercept
在测试数据上评估模型
test_p = lr_model.evaluate(test_df)
test_p.r2
test_p.meanSquaredError
五、预测数值
#预测
predictions = lr_model.transform(test_df)
print (predictions.collect())
print (predictions.show())
