一、下载包
settings -> interpreter ->
+ joblib 存取模型
+ matplotlib
+ numpy
+ pyspark
+ scikit-learn
二 、先确定pyCharm能用spark.sql连接hive成功
见 https://www.cnblogs.com/sabertobih/p/14183397.html
三、Kmeans目的：使用train集给events分label
（1） 确定k值
注意，其中的features字段在用工具执行kafka->hbase的过程中将_c0,_c1,_c2...多列合并成了一列string
而Kmeans中的VectorAssembler需要列名的Array集合
import numpy as np from matplotlib import pyplot as plt from pyspark.sql import SparkSession from pyspark.ml.clustering import KMeans from pyspark.sql.functions import split from pyspark.sql.types import DoubleType from pyspark.ml.feature import VectorAssembler if __name__ == '__main__': spark=SparkSession.builder.appName("test") .master("local[*]") .enableHiveSupport().getOrCreate() df = spark.sql("select eventid,features from dm_events.dm_usereventfinal").cache() #先从dataframe中将指定列进行分割 并获得返回新的dataframe spno 101列 和原来的df 一一对应 spno = split(df["features"],",") # 将101列的值通过withcolumn都补充到df上 同时将所有的列都转为double类型 for i in range(101): df = df.withColumn("c_"+str(i),spno.getItem(i).cast(DoubleType())) # 删除不需要的列 features df = df.drop("features") # 使用VectorAssembler 将c_0~c_100都组装为feature的向量列 ass = VectorAssembler(inputCols=["c_{0}".format(i) for i in range(101)],outputCol="feature") re = ass.transform(df) # 只保留需要的两列 re = re.select("eventid","feature") # 准备一个点数组 points = [i for i in range(2,50)] # 定义一个距离数组 distance=[] for po in points: print("point...........",po," ") # 使用kmeans指定特征列为feature clf = KMeans(k=po, featuresCol="feature") # 计算模型 model = clf.fit(re) # 计算所有到质心的平均距离 并存放到距离数组中 distance.append(model.computeCost(re)) # 绘制对应的折线图 plt.plot(points,distance,color="red") plt.show()
结果：
（2） 使用k训练模型
import numpy as np from matplotlib import pyplot as plt from pyspark.sql import SparkSession from pyspark.ml.clustering import KMeans from pyspark.sql.functions import split from pyspark.sql.types import DoubleType from pyspark.ml.feature import VectorAssembler from pyspark.ml.clustering import KMeans,KMeansModel if __name__ == '__main__': spark=SparkSession.builder.appName("test") .master("local[*]") .enableHiveSupport().getOrCreate() df = spark.sql("select eventid,features from dm_events.dm_usereventfinal").cache() #先从dataframe中将指定列进行分割 并获得返回新的dataframe spno 101列 和原来的df 一一对应 spno = split(df["features"],",") # 将101列的值通过withcolumn都补充到df上 同时将所有的列都转为double类型 for i in range(101): df = df.withColumn("c_"+str(i),spno.getItem(i).cast(DoubleType())) # 删除不需要的列 features df = df.drop("features") # 使用VectorAssembler 将c_0~c_100都组装为feature的向量列 ass = VectorAssembler(inputCols=["c_{0}".format(i) for i in range(101)],outputCol="feature") re = ass.transform(df) # 只保留需要的两列 re = re.select("eventid","feature") # 准备一个点数组 # points = [i for i in range(2,50)] # # 定义一个距离数组 # distance=[] # for po in points: # print("point...........",po," ") # # 使用kmeans指定特征列为feature # clf = KMeans(k=po, featuresCol="feature") # # 计算模型 # model = clf.fit(re) # # 计算所有到质心的平均距离 并存放到距离数组中 # distance.append(model.computeCost(re)) # # # 绘制对应的折线图 # plt.plot(points,distance,color="red") # plt.show() # 根据图形 得到最优的K值为10 clf = KMeans(k=10, featuresCol="feature") model = clf.fit(re) ## 训练成一个通用model，存起来以备复用 #### 模型存储方法1：上传hdfs KMeansModel.write(model).overwrite().save("hdfs://192.168.56.115:9000/party/model") #### 模型存储方法2：joblib #save model joblib.dump(clf,'../../data/model/randomforest.pkl',compress=3) #load model to clf clf = joblib.load('../../data/model/randomforest.pkl')
（3）模型读取，并将结果写入hive表中
# 读取模型 对数据进行分组 分组后存放到hive中 # transform是给model传参数，返回id，features，prediction => 直接drop掉向量列 dff = KMeansModel.load("hdfs://192.168.56.115:9000/party/model") .transform(re).select("eventid", "prediction") dff.write.mode("overwrite").saveAsTable("dm_events.dm_eventType")
四、RF目的：预测test集的label(if interested?)
from pyspark.sql import SparkSession from pyspark.ml.feature import VectorAssembler from pyspark.sql.types import DoubleType from pyspark.ml.classification import RandomForestClassifier,RandomForestClassificationModel from pyspark.ml.evaluation import BinaryClassificationEvaluator if __name__ == '__main__': spark = SparkSession.builder.master("local[*]") .appName("test").enableHiveSupport().getOrCreate() df = spark.sql("select * from dm_events.dm_final").distinct().cache() ## 1.选取作为features的列 cols = df.columns cols.remove("label") cols.remove("userid") cols.remove("eventid") cols.remove("age") # 该字段为null，不能变成向量 ## 2.预features列转成DoubleType for c in cols: df = df.withColumn(c,df[c].cast(DoubleType())) ## 3.预features列转成向量,并加入df中 ass = VectorAssembler(inputCols=cols,outputCol="features") ## df中加入向量列，取出features+label df= ass.transform(df) model_df = df.select(["features","label"]) .withColumn("label",df["label"].cast(DoubleType())) # 4.将df 2 8分开 train_df,test_df = model_df.randomSplit([0.8,0.2],seed=1225) # 5.训练模型，用RF训练 rf = RandomForestClassifier(labelCol="label",numTrees=128,maxDepth=9) model = rf.fit(train_df) # 用train结果生成模型，用fit # 6.存储模型 RandomForestClassifier.write(model).overwrite().save("d:/rfc")
# 7.读取模型，并测试准确性：0.7896044629820028 mod = RandomForestClassificationModel.load("d:/rfc") res = mod.transform(test_df) # 读取模型用transform，用来获得test结果 rfc_auc = BinaryClassificationEvaluator(labelCol="label").evaluate(res) print(rfc_auc) spark.stop()