Hadoop大数据平台构建与应用(工作手册式)（微课版） 案例 10.2 超市客户聚类分析

超市客户聚类分析【任务描述】在现代市场经济中，由于现代企业资源的有限性和消费需求的多样性，对客户进行必要的分类是非常重要的，这关系到未来营销战略的成败。目前很多大中型企业已经意识到了客户分类的重要性，开始寻求大数据的相关方法来解决客户分类问题。因为客户的需求具有异质性，即不是所有客户的需求都相同。由于客户需求、欲望及购买行为是多元的，所以客户需求满足呈现差异。对客户进行细分，可以让市场营销、销售人员以及企业的决策层从一个比较高的层次来观察客户信息数据仓库中的客户信息，使得企业可以针对不同类型的客户采用相应的营销策略，使企业市场营销服务活动的目标性和有效性得到提高，从而相对降低营销成本，最大限度的开发和维护客户资源，使企业的长期利润和持续发展得到保证。本任务将运用SparkSQL对超市客户数据进行探索分析，同时使用SparkMLlib提供的K-means算法构建聚类模型，最终实现对超市客户的聚类。【任务分析】超市客户的聚类分析的实现步骤如下。Scala插件安装，包括在线安装与离线安装介绍。配置Spark运行环境。对超市客户数据进行数据探索，包括性别分布统计、年龄分布统计，然后对数据进行特征构建，包括构建年收入等级特征、构建消费等级特征。将构建好特征的数据保存至Hive数据库中。使用K-means算法构建聚类模型，实现对超市客户的聚类。【任务实施】Scala插件安装在项目3，因为安装IntelliJIDEA时选择了默认安装，所以Scala插件并没有安装。Spark是由Scala语言编写而成的，因此IntelliJIDEA还需要安装Scala插件，配置Scala开发环境。Scala插件的安装有在线安装和离线安装两种方式，具体操作过程如下。在线安装Scala插件在线安装Scala插件的操作步骤如下。（1）打开IntelliJIDEA，单击界面右下角的“Configure”下拉列表，选择“Plugins”选项，如REF_Ref121755684\h图101所示。图STYLEREF1\s10SEQ图\*ARABIC\s11Plugins安装（2）弹图102。图STYLEREF1\s10SEQ图\*ARABIC\s12Plugins对话框（3）下载完成后，单击“RestartIDE”按钮重启IDEA，如REF_Ref121755697\h图103所示。图STYLEREF1\s10SEQ图\*ARABIC\s13Scala插件在线安装离线安装Scala插件使用离线安装的方式安装Scala插件时，Scala插件需要提前下载至本机计算机中。本书的IntelliJIDEA版本使用的Scala插件为“scala-intellij-bin-2018.3.6.zip”，可从IDEA官网下载。离线安装Scala插件的操作步骤如下。（1）在REF_Ref121755691\h图102所示的“Plugins”对话框中，单击图标，在下拉列表中选择“Installpluginfromdisk...”选项，弹出REF_Ref121755719\h图104所示的界面，选择Scala插件所在路径，单击“OK”按钮进行安装。图STYLEREF1\s10SEQ图\*ARABIC\s14Scala插件离线安装（2）Scala插件安装完成后将出现REF_Ref121755728\h图105所示的界面，单击右侧的“RestartIDE”按钮重启IntelliJIDEA即可。图STYLEREF1\s10SEQ图\*ARABIC\s15Scala插件安装完成配置Spark运行环境在IDEA中配置Spark运行环境，即需要在IDEA创建好的Scala工程中添加Spark开发依赖包。创建Maven工程参考项目2，创建名称定义为“MallCustomer_K”的Maven工程，MallCustomer_K工程页面如REF_Ref121759722\h图106所示。图STYLEREF1\s10SEQ图\*ARABIC\s16MallCustomer_K工程页面选择菜单栏中的“File”→“ProjectStructure”选项，打开REF_Ref121759727\h图107所示的界面。也可以使用“Ctrl+Alt+Shift+S”快捷键打开。图STYLEREF1\s10SEQ图\*ARABIC\s17工程结构打开REF_Ref121759727\h图107所示的工程结构配置界面后，选择“Libraries”选项，单击“+”按钮，选择“ScalaSDK”选项，在弹出的界面中选中2.12+版本的Scala插件单击“OK”，如REF_Ref121759740\h图108所示，单击“OK”按钮即可将Scala插件加入到工程中。图STYLEREF1\s10SEQ图\*ARABIC\s18添加Scala插件将Hive安装目录下的/conf/hive-site.xml、Hadoop安装目录下的/etc/hadoop/core-site.xml和hdfs-site.xml下载、复制到resource文件夹中，如REF_Ref121759745\h图109所示。图STYLEREF1\s10SEQ图\*ARABIC\s19添加配置文件添加Spark开发依赖包选择菜单栏中的“File”→“ProjectStructure”选项，打开REF_Ref121755776\h图1010所示的界面。也可以使用“Ctrl+Alt+Shift+S”快捷键打开。图STYLEREF1\s10SEQ图\*ARABIC\s110工程结构打开REF_Ref121755776\h图1010所示的工程结构配置界面后，选择“Libraries”选项，单击“+”按钮，选择“Java”选项，在弹出的界面中找到Spark安装目录下的jars文件夹，将整个文件夹导入，如REF_Ref121755784\h图1011所示，单击“OK”按钮即可将Spark开发依赖包加入到工程中。至此Spark的编程环境配置完成。图STYLEREF1\s10SEQ图\*ARABIC\s111添加Spark开发依赖包数据探索与特征构建超市客户数据集包含某超市通过会员卡获得一些客户的基本数据，包括客户ID、年龄、性别、年收入和消费分数。其中消费分数是根据定义的参数（例如客户行为和购买数据）分配给客户的，字段说明如REF_Ref99110210\h表101所示。表STYLEREF1\s10SEQ表\*ARABIC\s11数据集变量说明变量名变量说明变量类型及说明CustomerID客户的唯一ID编码IntGender客户性别StringAge客户年龄IntAnnualIncome客户的年收入Int，单位(k$)SpendingScore消费分数Int，范围(1-100)读取超市客户数据，并创建DataFrame。由于数据比较多，将数据集文件上传至HDFS文件系统的/user/root目录下，同时在Hive中创建名为“mall”的数据库。如REF_Ref99114440\h代码101所示。本案例的完整流程都将在IDEA开发环境中通过编程实现，具体步骤如下。代码STYLEREF1\s10SEQ代码\*ARABIC\s11将数据上传至HDFS//将数据上传至HDFShdfsdfs-put/data/Mall_Customers.csv/user/root///在Hive中创建名为“mall”的数据库createdatabasemall;（1）配置SparkSession并读取数据右键单击MallCustomer_K工程下的src/main/java文件夹，依次选择“New”→“ScalaClass”选项，在包下新建一个Scala类，将Scala类的类名设置为“customerDataAnalyse”，并在“Kind”右侧的下拉列表中选择“Object”选项，单击“OK”按钮，完成Scala类的创建。配置SparkSession，从HDFS中读取超市客户数据为DataFrame，并查看DataFrame的前6行数据，如REF_Ref99370330\h代码102所示，返回结果如REF_Ref99370468\h图1012所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s12配置SparkSession并读取数据//配置SparkSession

valspark=SparkSession.builder()

.master("local[*]")

.appName("customerDataAnalyse")

.enableHiveSupport()

.getOrCreate()

spark.sparkContext.setLogLevel("WARN")

//读取超市客户数据

valdata=spark.read

.option("header","true")

.option("inferSchema","true")

.option("delimiter",",")

.csv("hdfs://master:8020/user/root/Mall_Customers.csv")

//查看DataFrame的前6行数据

data.show(6,false)图STYLEREF1\s10SEQ图\*ARABIC\s112查看前6行数据（2）性别分布统计由于男女性在对待购物这件事情上看法可能会有所差别，所以先查看不同性别的客户人数以及占比情况，如REF_Ref99377197\h代码103所示，返回结果如REF_Ref99377206\h图1013所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s13性别分布统计importorg.apache.spark.sql.functions._

println("性别分布统计,统计各性别人数，计算占比情况：")

data.groupBy("Gender").count()

.withColumn("GenderPercent",round(col("count")/data.count()*100,2))

.show(false)图STYLEREF1\s10SEQ图\*ARABIC\s113性别分布统计（3）年龄分布统计不同年龄的人在对待购物这件事情上的看法也可能有所差别，因此需要查看客户年龄的分布情况，如REF_Ref99377848\h代码104所示，返回结果如REF_Ref99377862\h图1014所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s14年龄分布统计println("年龄分布统计，统计各年龄人数，计算占比情况：")

data.groupBy("Age").count()

.withColumn("AgePercent",round(col("count")/data.count(),2))

.sort(desc("count"))

.show(false)图STYLEREF1\s10SEQ图\*ARABIC\s114年龄分布统计（4）构建年收入等级特征客户的年收入水平同样是一个显著影响客户消费的指标，根据客户的年收入构建年收入等级特征，作为新的一列保存至新的DataFrame中，并查看分布情况，如REF_Ref99380727\h代码105所示，返回结果如REF_Ref99380740\h图1015所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s15构建年收入等级特征//年收入等级划分，根据收入字段划分并新建收入水平字段，//<35，35~56，56~77，77~98，>98，划分为5个等级，并计算各等级数量和占比情况

println("年收入等级划分：")

valnewdata1=data.withColumn("IncomeLevel",

when(col("AnnualIncome(k$)")<35,1).

when(col("AnnualIncome(k$)")>35andcol("AnnualIncome(k$)")<=56,2).

when(col("AnnualIncome(k$)")>56andcol("AnnualIncome(k$)")<=77,3).

when(col("AnnualIncome(k$)")>77andcol("AnnualIncome(k$)")<=98,4).

when(col("AnnualIncome(k$)")>98,5).

otherwise(6))

newdata1.groupBy("IncomeLevel").count().withColumn("IncomeLevelPercent",round(col("count")/data.count()*100,2))

.show()图STYLEREF1\s10SEQ图\*ARABIC\s115年收入等级统计（5）构建消费等级特征消费分数代表着客户对于商城的贡献和价值，也可以侧面反映出客户对于此商城的满意程度。根据用户的消费分数构建客户消费等级特征，将其作为新的一列保存至新的DataFrame中，并查看分布情况，如REF_Ref99381589\h代码106所示，返回结果如REF_Ref99382517\h图1016所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s16构建消费等级特征//消费得分分布，根据消费字段新建消费水平等级字段，//<10，10~30，30~50，50~70，70~100，划分为5个等级，并计算各个等级的数量和占比情况

println("消费等级分布：")

valnewdata2=newdata1.withColumn("SpendLevel",

when(col("SpendingScore(1-100)")<=10,1).

when(col("SpendingScore(1-100)")>10andcol("SpendingScore(1-100)")<=30,2).

when(col("SpendingScore(1-100)")>30andcol("SpendingScore(1-100)")<=50,3).

when(col("SpendingScore(1-100)")>50andcol("SpendingScore(1-100)")<=70,4).

when(col("SpendingScore(1-100)")>70andcol("SpendingScore(1-100)")<=100,5))

newdata2.groupBy("SpendLevel").count().withColumn("SpendLevelPercent",round(col("count")/data.count()*100,2))

.show()图STYLEREF1\s10SEQ图\*ARABIC\s116消费等级分布统计保存数据至Hive在Xshell中输入命令启动Hive的metastore服务以及hiveserver2服务，如REF_Ref99393569\h代码107所示。对客户年收入与消费分数两个字段进行重命名，方便后续机器学习的模型构建，并将修改后的客户数据写入至Hive中，如REF_Ref99396951\h代码108所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s17启动Hive服务//启动metastore服务与hiveserver2服务hive--servicemetastore&hive--servicehiveserver2&代码STYLEREF1\s10SEQ代码\*ARABIC\s18重命名字段并保存数据至Hive//重命名字段println("对字段名进行重命名：")

valnewdata3=newdata2.withColumnRenamed("AnnualIncome(k$)","Income")

.withColumnRenamed("SpendingScore(1-100)","Spending")//写入数据至Hivenewdata3.write.mode("overwrite")

.option("header","true")

.saveAsTable("mall.customer")

println("写入数据至Hive表...成功！")在Hive中查看已导入的数据，如REF_Ref99397206\h图1017所示。图STYLEREF1\s10SEQ图\*ARABIC\s117查看customer表K-means聚类在MallCustomer_K工程下的src/main/java文件夹下创建类名为“kmeansModel”的Scala类，“Kind”选择“Object”选项。在构建K-means聚类模型前，需要将建模所需的特征字段类型修改为Double类型，使用StringIndexer将客户性别（Gender）字段的值编码成标签索引，再使用VectorAssembler将用户特征合并到一个特征列中。构建K-means模型参数网格，设置最大迭代次数、随机种子和k值，使用训练集切分验证，获取最佳参数模型，设置模型评估器为ClusteringEvaluator，训练模型，并得到聚类结果。完整实现如REF_Ref99401155\h代码109所示，返回结果如REF_Ref99401695\h图1018、REF_Ref119511141\h图1019、REF_Ref99401711\h图1020所示。代码STYLEREF1\s10SEQ代码\*ARABIC\s19K-means聚类importorg.apache.spark.ml.Pipeline

importorg.apache.spark.ml.clustering.KMeans

importorg.apache.spark.ml.evaluation.ClusteringEvaluator

importorg.apache.spark.ml.feature.{StringIndexer,VectorAssembler}

importorg.apache.spark.ml.tuning.{ParamGridBuilder,TrainValidationSplit}

importorg.apache.spark.sql.SparkSession

importorg.apache.spark.sql.functions.desc

objectkmeansModel{

defmain(args:Array[String]):Unit={

valspark=SparkSession.builder()

.appName("kmeansModel")

.master("local[*]")

.enableHiveSupport()

.getOrCreate()

spark.sparkContext.setLogLevel("WARN")

/**

*修改需要进行模型计算的特征字段为double类型

*/

valdata=spark.table("mall.customer")

.selectExpr("cast(ageasdouble)age",

"cast(incomeasdouble)income",

"cast(spendingasdouble)spending",

"gender")

//获取数据列名

valcolumns=data.columns

/**

*构建模型，包含StringIndex（将gender性别字符数据转换为Int类型）、VectorAssembler（整合特征字段为Vector）

*和Kmeans模型。将词三个模型放入到管道中

*/

valStringIndex=newStringIndexer()

.setInputCol("gender")

.setOutputCol("sex")

valVector=newVectorAssembler()

.setInputCols((columns:+"sex").filter(!_.contains("gender")))

.setOutputCol("features")

valKMeanModel=newKMeans()

valpipeline=newPipeline()

.setStages(Array(StringIndex,Vector,KMeanModel))

/**

*构建Kmeans模型参数网格，设置最大迭代次数，随机种子和k值

*使用训练集切分验证（默认75%），获取最佳参数模型，设置模型评估器为ClusteringEvaluator

*训练模型，并得到聚类结果

*/

valparamGrid=newParamGridBuilder()

.addGrid(KMeanModel.maxIter,Array(10,20,50))

.addGrid(KMeanModel.seed,Array(1L,2L,3L))

.addGrid(KMeanModel.k,Array(3,4,5,6,7,8,9,10))

.build()

valtrainValidationSplit=newTrainValidationSplit()

.setEstimator(pipeline)

.setEvaluator(newClusteringEvaluator())

.setEstimatorParamMaps(param