SparkMLlib Kmeans客户细分-白红宇

SparkMLlib Kmeans客户细分

阅读量：5866 次

发布时间：2019-06-19

本文共 7993 字，大约阅读时间需要 26 分钟。

第一章概述

1.1 Kmeans原理

首先区分分类与聚类，其中输入数据拥有类别标签，通过对已知类别的训练，找到不同类别的数据特性从而形成分类模型。再使用模型对未分类的数据进行分类的属于分类。而输入数据没有类别区分，直接通过聚类算法将数据聚合为多个群组的属于聚类。

kmeans作为一种基础的聚类算法，在客户细分等场景中有广泛应用，其算法过程大致分为以下四步：

第一步，选择 K 个点作为初始聚类中心。（K值选择可参考肘部原则）

第二步，计算其余所有点到聚类中心的距离，并把每个点划分到离它最近的聚类中心所在的聚类中去。

第三步，计算每个聚类中所有点的算术平均值，并将其作为新的聚类中心点。

第四部，重复2，3步骤，直到函数收敛到一个相对稳定的值

1.2 实例分析

我们的应用场景是将酒店顾客进行细分，从而明确用户团组的特征，采取不同措施进行营销活动。

选定四个数据：最近消费时间、订单量、订单金额、折扣率

最近消费：最近一次消费离现在的时间

订单总量：范围时间内的订单量

订单金额：范围时间内的订单金额

折扣率：总底价/总卖价的比率，用以表示利润率

第二章数据抽取分析

2.1 数据抽取

create table testa asSELECT  user_no,  max(op_date)                                           last_date,  count(*)                                               order_cnt,  cast(sum(sum_price) AS DECIMAL(20, 2))                 sum_price,  cast(sum(sum_sign) / sum(sum_price) AS DECIMAL(10, 4)) sale_rateFROM orderWHERE month IN ('201807', '201808', '201809')GROUP BY user_no;

2.2 数据分析

用于查找无效数据、孤立数据、噪声数据，避免异常数据对最终结果的不良影响。

--order_cntSELECT  'order_cnt'       col,  avg(order_cnt)    avg,  stddev(order_cnt) stddev,  min(order_cnt)    min ,  percentile(order_cnt,0.25) per1,  percentile(order_cnt,0.5)  per2,  percentile(order_cnt,0.75) per3,  max(order_cnt)    maxFROM testa--sum_priceunionSELECT  'sum_price'       col,  avg(sum_price)    avg,  stddev(sum_price) stddev,  min(sum_price)    min ,  percentile(sum_price,0.25) per1,  percentile(sum_price,0.5)  per2,  percentile(sum_price,0.75) per3,  max(sum_price)    maxFROM testa--sale_rateunionSELECT  'sale_rate'       col,  avg(sale_rate)    avg,  stddev(sale_rate) stddev,  min(sale_rate)    min ,  percentile(sale_rate,0.25) per1,  percentile(sale_rate,0.5)  per2,  percentile(sale_rate,0.75) per3,  max(sale_rate)    maxFROM testa

分析结果：

	avg	标准差	min	0.25	0.5	0.75	max
order_cnt	2.86	594.44	1	1.0	1.0	2.0	940655
sum_price	1082.76	306789.61	0	166.0	302.99	664.0	522843380.09
sale_rate	0.8937	0.0412	-0.001	0.8619	0.8924	0.9036	7.0881

第三章数据预处理

3.1 数据清洗

通过对比每一列均值、中位数、最大最小等值，可以看书该列是否存在明显的噪声数据，剔除明显偏离的噪声可避免训练结果被噪声影响。

CREATE TABLE testb AS  SELECT    user_no,    last_date,    order_cnt,    sum_price,    sale_rate  FROM testa  WHERE order_cnt <= 50        AND sum_price < 10000        AND sale_rate >= 0.7 AND sale_rate < 1

数据范例：

user_no      last_date       order_cnt sum_price  sale_rate664986  2018-07-03 21:21:15.0    1       1116     0.87748612  2018-09-27 22:27:31.0    1       132      0.8788295402  2018-07-31 17:11:01.0    1       198      0.9

3.2 数据转换

将数据转换为适当格式，以满足后续模型训练的需要。

create table testc asSELECT  user_no,  datediff('2018-10-01', last_date)    last_date,  order_cnt,  sum_price,  sale_rateFROM testbWHERE order_cnt<=50and sum_price<10000and sale_rate>=0.7 and sale_rate<1

数据范例：

user_no last_date order_cnt sum_price  sale_rate664986  90        1         1116       0.87748612  4         1         132        0.8788295402  62        1         198        0.9

3.3 规范化

规范化：为消除不同数据列量级对聚类结果的不同程度影响，需对数据执行规范化。常用的数据规范化的方法有：正则化、标准化、归一化等，这里选择标准化方法。

正则化：原数据 /（列1^2+列2^2...+列N^2）^(1/2)

标准化：（原数据-平均值）/ 标准差

归一化：（原数据-最小值）/（最大值-最小值）

--计算每一列平均值，标准差SELECT  cast(avg(last_date) AS DECIMAL(20, 4)),  cast(stddev(last_date) AS DECIMAL(20, 4)),  cast(avg(order_cnt) AS DECIMAL(20, 4)),  cast(stddev(order_cnt) AS DECIMAL(20, 4)),  cast(avg(sum_price) AS DECIMAL(20, 4)),  cast(stddev(sum_price) AS DECIMAL(20, 4)),  cast(avg(sale_rate) AS DECIMAL(20, 4)),  cast(stddev(sale_rate) AS DECIMAL(20, 4))FROM testc

执行标准化：

create table testd asSELECT  user_no,  cast((last_date-42.6495 )/26.2554 AS DECIMAL(10, 4)) last_date,  cast((order_cnt-2.1737)/2.3788 AS DECIMAL(10, 4))    order_cnt,  cast((sum_price-654.6573)/1028.437 AS DECIMAL(10, 4)) sum_price,  cast((sale_rate-0.8911)/0.0376 AS DECIMAL(10, 4))     sale_rateFROM testc

数据范例：

user_no last_date  order_cnt sum_price  sale_rate664986   1.8035    -0.4934    0.4486    -0.5612748612  -1.4721    -0.4934   -0.5082    -0.3271295402   0.737     -0.4934.  -0.444      0.2367

第四章模型构建

4.1 K值选取

Spark MLlib 在 KMeansModel 类里提供了 computeCost 方法，该方法通过计算所有数据点到其最近的中心点的平方和来评估聚类的效果；通常利用肘部原则选择合适K值，即随K增大误差平方和明显收敛的点作为K值。此处我们选择4为K值。

//使用误差平方之和来评估数据模型val cost = model.computeCost(parsedData)

4.2 模型构建

模型生成：

import org.apache.log4j.{Level, Logger}import org.apache.spark.{SparkConf, SparkContext}import org.apache.spark.mllib.clustering.KMeansimport org.apache.spark.mllib.linalg.Vectorsobject ml_kmeans_user {  def main(args: Array[String]) {    // 屏蔽不必要的日志显示在终端上    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)    // 设置运行环境    val conf = new SparkConf().setAppName("Kmeans_user").setMaster("local[4]")    val sc = new SparkContext(conf)    // 装载数据集，例：690780,-1.5863,2.4493,1.8993,-0.5532    val data = sc.textFile("/Users/user/testscala/ml/kmeans_user/kmeans_card_no.csv", 1)    val rate_data = data.map(x=>    {val y=x.split(',')      y(1)+","+y(2)+","+y(3)+","+y(4)    })    val parsedData=rate_data.map(s => Vectors.dense(s.split(',').map(_.toDouble)))    // 将数据集聚类，4个类，20次迭代，进行模型训练形成数据模型    val numClusters = 4    val numIterations = 20    val model = KMeans.train(parsedData, numClusters, numIterations)    // 打印数据模型的中心点    println("Cluster centers:")    for (c <- model.clusterCenters) {      println("  " + c.toString)    }    // 使用误差平方之和来评估数据模型    val cost = model.computeCost(parsedData)    println("Within Set Sum of Squared Errors = " + cost)    // 使用模型测试单点数据    println("Vectors -1.0150,0.7677,4.1931,-0.5957 is belongs to clusters:"      + model.predict(Vectors.dense("-1.0150,0.7677,4.1931,-0.5957".split(',').map(_.toDouble))))    println("Vectors 0.6227,-0.4934,0.0125,-0.6702 is belongs to clusters:"      + model.predict(Vectors.dense("0.6227,-0.4934,0.0125,-0.6702".split(',').map(_.toDouble))))    println("Vectors 0.4615,0.0279,0.0132,0.3333 is belongs to clusters:"      + model.predict(Vectors.dense("-1.1674,-0.4934,0.5060,0.0000".split(',').map(_.toDouble))))    // 交叉评估1，只返回结果    val testdata = rate_data.map(s => Vectors.dense(s.split(',').map(_.toDouble)))    val result1 = model.predict(testdata)    result1.saveAsTextFile("/Users/user/testscala/ml/kmeans_user/result_kmeans1")    //输出每种分类的数量和比例    val rows_num=result1.count    println("Group's rows and rate:")    result1.countByValue().map(x=>(x._1,x._2,rows_num,(x._2.toDouble/rows_num))).foreach(println)    // 交叉评估2，返回数据集和结果    val result2 = data.map {line =>      val x=line.split(',')      val y= (x(1)+","+x(2)+","+x(3)+","+x(4))  //去除行首的酒店ID      val linevectore = Vectors.dense(y.split(',').map(_.toDouble))      val prediction = model.predict(linevectore)      line + " " + prediction    }.saveAsTextFile("/Users/user/testscala/ml/kmeans_user/result_kmeans2")    sc.stop()  }}

模型输出：

Cluster centers:  [-0.8541582382496996,-0.08279057683953428,-0.15476461980572706,-0.2580014912415115]  [-0.5770962398956121,2.603990288650971,2.9302383887085415,-0.011702418687543834]  [0.25562458161693147,-0.14200265492890868,-0.23666049251057267,1.8315037304226196]  [0.8657341495823884,-0.27286596904883526,-0.2148085017552938,-0.4060542342028446]Within Set Sum of Squared Errors = 1799254.8630008646Vectors -1.0150,0.7677,4.1931,-0.5957 is belongs to clusters:1Vectors 0.6227,-0.4934,0.0125,-0.6702 is belongs to clusters:3Vectors 0.4615,0.0279,0.0132,0.3333 is belongs to clusters:0Group's rows and rate:(0,400883,1000000,0.400883)(1,62851,1000000,0.062851)(2,143468,1000000,0.143468)(3,392798,1000000,0.392798)