python3 通过pyspark抽取Hive数据 进行线性回

背景：抽取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())

---