Hadoop大數(shù)據(jù)平臺(tái)構(gòu)建與應(yīng)用(工作手冊(cè)式)（微課版） 案例 10.2 超市客戶(hù)聚類(lèi)分析

超市客戶(hù)聚類(lèi)分析【任務(wù)描述】在現(xiàn)代市場(chǎng)經(jīng)濟(jì)中，由于現(xiàn)代企業(yè)資源的有限性和消費(fèi)需求的多樣性，對(duì)客戶(hù)進(jìn)行必要的分類(lèi)是非常重要的，這關(guān)系到未來(lái)營(yíng)銷(xiāo)戰(zhàn)略的成敗。目前很多大中型企業(yè)已經(jīng)意識(shí)到了客戶(hù)分類(lèi)的重要性，開(kāi)始尋求大數(shù)據(jù)的相關(guān)方法來(lái)解決客戶(hù)分類(lèi)問(wèn)題。因?yàn)榭蛻?hù)的需求具有異質(zhì)性，即不是所有客戶(hù)的需求都相同。由于客戶(hù)需求、欲望及購(gòu)買(mǎi)行為是多元的，所以客戶(hù)需求滿(mǎn)足呈現(xiàn)差異。對(duì)客戶(hù)進(jìn)行細(xì)分，可以讓市場(chǎng)營(yíng)銷(xiāo)、銷(xiāo)售人員以及企業(yè)的決策層從一個(gè)比較高的層次來(lái)觀察客戶(hù)信息數(shù)據(jù)倉(cāng)庫(kù)中的客戶(hù)信息，使得企業(yè)可以針對(duì)不同類(lèi)型的客戶(hù)采用相應(yīng)的營(yíng)銷(xiāo)策略，使企業(yè)市場(chǎng)營(yíng)銷(xiāo)服務(wù)活動(dòng)的目標(biāo)性和有效性得到提高，從而相對(duì)降低營(yíng)銷(xiāo)成本，最大限度的開(kāi)發(fā)和維護(hù)客戶(hù)資源，使企業(yè)的長(zhǎng)期利潤(rùn)和持續(xù)發(fā)展得到保證。本任務(wù)將運(yùn)用SparkSQL對(duì)超市客戶(hù)數(shù)據(jù)進(jìn)行探索分析，同時(shí)使用SparkMLlib提供的K-means算法構(gòu)建聚類(lèi)模型，最終實(shí)現(xiàn)對(duì)超市客戶(hù)的聚類(lèi)。【任務(wù)分析】超市客戶(hù)的聚類(lèi)分析的實(shí)現(xiàn)步驟如下。Scala插件安裝，包括在線(xiàn)安裝與離線(xiàn)安裝介紹。配置Spark運(yùn)行環(huán)境。對(duì)超市客戶(hù)數(shù)據(jù)進(jìn)行數(shù)據(jù)探索，包括性別分布統(tǒng)計(jì)、年齡分布統(tǒng)計(jì)，然后對(duì)數(shù)據(jù)進(jìn)行特征構(gòu)建，包括構(gòu)建年收入等級(jí)特征、構(gòu)建消費(fèi)等級(jí)特征。將構(gòu)建好特征的數(shù)據(jù)保存至Hive數(shù)據(jù)庫(kù)中。使用K-means算法構(gòu)建聚類(lèi)模型，實(shí)現(xiàn)對(duì)超市客戶(hù)的聚類(lèi)?！救蝿?wù)實(shí)施】Scala插件安裝在項(xiàng)目3，因?yàn)榘惭bIntelliJIDEA時(shí)選擇了默認(rèn)安裝，所以Scala插件并沒(méi)有安裝。Spark是由Scala語(yǔ)言編寫(xiě)而成的，因此IntelliJIDEA還需要安裝Scala插件，配置Scala開(kāi)發(fā)環(huán)境。Scala插件的安裝有在線(xiàn)安裝和離線(xiàn)安裝兩種方式，具體操作過(guò)程如下。在線(xiàn)安裝Scala插件在線(xiàn)安裝Scala插件的操作步驟如下。（1）打開(kāi)IntelliJIDEA，單擊界面右下角的“Configure”下拉列表，選擇“Plugins”選項(xiàng)，如REF_Ref121755684\h圖101所示。圖STYLEREF1\s10SEQ圖\*ARABIC\s11Plugins安裝（2）彈圖102。圖STYLEREF1\s10SEQ圖\*ARABIC\s12Plugins對(duì)話(huà)框（3）下載完成后，單擊“RestartIDE”按鈕重啟IDEA，如REF_Ref121755697\h圖103所示。圖STYLEREF1\s10SEQ圖\*ARABIC\s13Scala插件在線(xiàn)安裝離線(xiàn)安裝Scala插件使用離線(xiàn)安裝的方式安裝Scala插件時(shí)，Scala插件需要提前下載至本機(jī)計(jì)算機(jī)中。本書(shū)的IntelliJIDEA版本使用的Scala插件為“scala-intellij-bin-2018.3.6.zip”，可從IDEA官網(wǎng)下載。離線(xiàn)安裝Scala插件的操作步驟如下。（1）在REF_Ref121755691\h圖102所示的“Plugins”對(duì)話(huà)框中，單擊圖標(biāo)，在下拉列表中選擇“Installpluginfromdisk...”選項(xiàng)，彈出REF_Ref121755719\h圖104所示的界面，選擇Scala插件所在路徑，單擊“OK”按鈕進(jìn)行安裝。圖STYLEREF1\s10SEQ圖\*ARABIC\s14Scala插件離線(xiàn)安裝（2）Scala插件安裝完成后將出現(xiàn)REF_Ref121755728\h圖105所示的界面，單擊右側(cè)的“RestartIDE”按鈕重啟IntelliJIDEA即可。圖STYLEREF1\s10SEQ圖\*ARABIC\s15Scala插件安裝完成配置Spark運(yùn)行環(huán)境在IDEA中配置Spark運(yùn)行環(huán)境，即需要在IDEA創(chuàng)建好的Scala工程中添加Spark開(kāi)發(fā)依賴(lài)包。創(chuàng)建Maven工程參考項(xiàng)目2，創(chuàng)建名稱(chēng)定義為“MallCustomer_K”的Maven工程，MallCustomer_K工程頁(yè)面如REF_Ref121759722\h圖106所示。圖STYLEREF1\s10SEQ圖\*ARABIC\s16MallCustomer_K工程頁(yè)面選擇菜單欄中的“File”→“ProjectStructure”選項(xiàng)，打開(kāi)REF_Ref121759727\h圖107所示的界面。也可以使用“Ctrl+Alt+Shift+S”快捷鍵打開(kāi)。圖STYLEREF1\s10SEQ圖\*ARABIC\s17工程結(jié)構(gòu)打開(kāi)REF_Ref121759727\h圖107所示的工程結(jié)構(gòu)配置界面后，選擇“Libraries”選項(xiàng)，單擊“+”按鈕，選擇“ScalaSDK”選項(xiàng)，在彈出的界面中選中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下載、復(fù)制到resource文件夾中，如REF_Ref121759745\h圖109所示。圖STYLEREF1\s10SEQ圖\*ARABIC\s19添加配置文件添加Spark開(kāi)發(fā)依賴(lài)包選擇菜單欄中的“File”→“ProjectStructure”選項(xiàng)，打開(kāi)REF_Ref121755776\h圖1010所示的界面。也可以使用“Ctrl+Alt+Shift+S”快捷鍵打開(kāi)。圖STYLEREF1\s10SEQ圖\*ARABIC\s110工程結(jié)構(gòu)打開(kāi)REF_Ref121755776\h圖1010所示的工程結(jié)構(gòu)配置界面后，選擇“Libraries”選項(xiàng)，單擊“+”按鈕，選擇“Java”選項(xiàng)，在彈出的界面中找到Spark安裝目錄下的jars文件夾，將整個(gè)文件夾導(dǎo)入，如REF_Ref121755784\h圖1011所示，單擊“OK”按鈕即可將Spark開(kāi)發(fā)依賴(lài)包加入到工程中。至此Spark的編程環(huán)境配置完成。圖STYLEREF1\s10SEQ圖\*ARABIC\s111添加Spark開(kāi)發(fā)依賴(lài)包數(shù)據(jù)探索與特征構(gòu)建超市客戶(hù)數(shù)據(jù)集包含某超市通過(guò)會(huì)員卡獲得一些客戶(hù)的基本數(shù)據(jù)，包括客戶(hù)ID、年齡、性別、年收入和消費(fèi)分?jǐn)?shù)。其中消費(fèi)分?jǐn)?shù)是根據(jù)定義的參數(shù)（例如客戶(hù)行為和購(gòu)買(mǎi)數(shù)據(jù)）分配給客戶(hù)的，字段說(shuō)明如REF_Ref99110210\h表101所示。表STYLEREF1\s10SEQ表\*ARABIC\s11數(shù)據(jù)集變量說(shuō)明變量名變量說(shuō)明變量類(lèi)型及說(shuō)明CustomerID客戶(hù)的唯一ID編碼IntGender客戶(hù)性別StringAge客戶(hù)年齡IntAnnualIncome客戶(hù)的年收入Int，單位(k$)SpendingScore消費(fèi)分?jǐn)?shù)Int，范圍(1-100)讀取超市客戶(hù)數(shù)據(jù)，并創(chuàng)建DataFrame。由于數(shù)據(jù)比較多，將數(shù)據(jù)集文件上傳至HDFS文件系統(tǒng)的/user/root目錄下，同時(shí)在Hive中創(chuàng)建名為“mall”的數(shù)據(jù)庫(kù)。如REF_Ref99114440\h代碼101所示。本案例的完整流程都將在IDEA開(kāi)發(fā)環(huán)境中通過(guò)編程實(shí)現(xiàn)，具體步驟如下。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s11將數(shù)據(jù)上傳至HDFS//將數(shù)據(jù)上傳至HDFShdfsdfs-put/data/Mall_Customers.csv/user/root///在Hive中創(chuàng)建名為“mall”的數(shù)據(jù)庫(kù)createdatabasemall;（1）配置SparkSession并讀取數(shù)據(jù)右鍵單擊MallCustomer_K工程下的src/main/java文件夾，依次選擇“New”→“ScalaClass”選項(xiàng)，在包下新建一個(gè)Scala類(lèi)，將Scala類(lèi)的類(lèi)名設(shè)置為“customerDataAnalyse”，并在“Kind”右側(cè)的下拉列表中選擇“Object”選項(xiàng)，單擊“OK”按鈕，完成Scala類(lèi)的創(chuàng)建。配置SparkSession，從HDFS中讀取超市客戶(hù)數(shù)據(jù)為DataFrame，并查看DataFrame的前6行數(shù)據(jù)，如REF_Ref99370330\h代碼102所示，返回結(jié)果如REF_Ref99370468\h圖1012所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s12配置SparkSession并讀取數(shù)據(jù)//配置SparkSession

valspark=SparkSession.builder()

.master("local[*]")

.appName("customerDataAnalyse")

.enableHiveSupport()

.getOrCreate()

spark.sparkContext.setLogLevel("WARN")

//讀取超市客戶(hù)數(shù)據(jù)

valdata=spark.read

.option("header","true")

.option("inferSchema","true")

.option("delimiter",",")

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

//查看DataFrame的前6行數(shù)據(jù)

data.show(6,false)圖STYLEREF1\s10SEQ圖\*ARABIC\s112查看前6行數(shù)據(jù)（2）性別分布統(tǒng)計(jì)由于男女性在對(duì)待購(gòu)物這件事情上看法可能會(huì)有所差別，所以先查看不同性別的客戶(hù)人數(shù)以及占比情況，如REF_Ref99377197\h代碼103所示，返回結(jié)果如REF_Ref99377206\h圖1013所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s13性別分布統(tǒng)計(jì)importorg.apache.spark.sql.functions._

println("性別分布統(tǒng)計(jì),統(tǒng)計(jì)各性別人數(shù)，計(jì)算占比情況：")

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

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

.show(false)圖STYLEREF1\s10SEQ圖\*ARABIC\s113性別分布統(tǒng)計(jì)（3）年齡分布統(tǒng)計(jì)不同年齡的人在對(duì)待購(gòu)物這件事情上的看法也可能有所差別，因此需要查看客戶(hù)年齡的分布情況，如REF_Ref99377848\h代碼104所示，返回結(jié)果如REF_Ref99377862\h圖1014所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s14年齡分布統(tǒng)計(jì)println("年齡分布統(tǒng)計(jì)，統(tǒng)計(jì)各年齡人數(shù)，計(jì)算占比情況：")

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

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

.sort(desc("count"))

.show(false)圖STYLEREF1\s10SEQ圖\*ARABIC\s114年齡分布統(tǒng)計(jì)（4）構(gòu)建年收入等級(jí)特征客戶(hù)的年收入水平同樣是一個(gè)顯著影響客戶(hù)消費(fèi)的指標(biāo)，根據(jù)客戶(hù)的年收入構(gòu)建年收入等級(jí)特征，作為新的一列保存至新的DataFrame中，并查看分布情況，如REF_Ref99380727\h代碼105所示，返回結(jié)果如REF_Ref99380740\h圖1015所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s15構(gòu)建年收入等級(jí)特征//年收入等級(jí)劃分，根據(jù)收入字段劃分并新建收入水平字段，//<35，35~56，56~77，77~98，>98，劃分為5個(gè)等級(jí)，并計(jì)算各等級(jí)數(shù)量和占比情況

println("年收入等級(jí)劃分：")

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年收入等級(jí)統(tǒng)計(jì)（5）構(gòu)建消費(fèi)等級(jí)特征消費(fèi)分?jǐn)?shù)代表著客戶(hù)對(duì)于商城的貢獻(xiàn)和價(jià)值，也可以側(cè)面反映出客戶(hù)對(duì)于此商城的滿(mǎn)意程度。根據(jù)用戶(hù)的消費(fèi)分?jǐn)?shù)構(gòu)建客戶(hù)消費(fèi)等級(jí)特征，將其作為新的一列保存至新的DataFrame中，并查看分布情況，如REF_Ref99381589\h代碼106所示，返回結(jié)果如REF_Ref99382517\h圖1016所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s16構(gòu)建消費(fèi)等級(jí)特征//消費(fèi)得分分布，根據(jù)消費(fèi)字段新建消費(fèi)水平等級(jí)字段，//<10，10~30，30~50，50~70，70~100，劃分為5個(gè)等級(jí)，并計(jì)算各個(gè)等級(jí)的數(shù)量和占比情況

println("消費(fèi)等級(jí)分布：")

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消費(fèi)等級(jí)分布統(tǒng)計(jì)保存數(shù)據(jù)至Hive在Xshell中輸入命令啟動(dòng)Hive的metastore服務(wù)以及hiveserver2服務(wù)，如REF_Ref99393569\h代碼107所示。對(duì)客戶(hù)年收入與消費(fèi)分?jǐn)?shù)兩個(gè)字段進(jìn)行重命名，方便后續(xù)機(jī)器學(xué)習(xí)的模型構(gòu)建，并將修改后的客戶(hù)數(shù)據(jù)寫(xiě)入至Hive中，如REF_Ref99396951\h代碼108所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s17啟動(dòng)Hive服務(wù)//啟動(dòng)metastore服務(wù)與hiveserver2服務(wù)hive--servicemetastore&hive--servicehiveserver2&代碼STYLEREF1\s10SEQ代碼\*ARABIC\s18重命名字段并保存數(shù)據(jù)至Hive//重命名字段println("對(duì)字段名進(jìn)行重命名：")

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

.withColumnRenamed("SpendingScore(1-100)","Spending")//寫(xiě)入數(shù)據(jù)至Hivenewdata3.write.mode("overwrite")

.option("header","true")

.saveAsTable("mall.customer")

println("寫(xiě)入數(shù)據(jù)至Hive表...成功！")在Hive中查看已導(dǎo)入的數(shù)據(jù)，如REF_Ref99397206\h圖1017所示。圖STYLEREF1\s10SEQ圖\*ARABIC\s117查看customer表K-means聚類(lèi)在MallCustomer_K工程下的src/main/java文件夾下創(chuàng)建類(lèi)名為“kmeansModel”的Scala類(lèi)，“Kind”選擇“Object”選項(xiàng)。在構(gòu)建K-means聚類(lèi)模型前，需要將建模所需的特征字段類(lèi)型修改為Double類(lèi)型，使用StringIndexer將客戶(hù)性別（Gender）字段的值編碼成標(biāo)簽索引，再使用VectorAssembler將用戶(hù)特征合并到一個(gè)特征列中。構(gòu)建K-means模型參數(shù)網(wǎng)格，設(shè)置最大迭代次數(shù)、隨機(jī)種子和k值，使用訓(xùn)練集切分驗(yàn)證，獲取最佳參數(shù)模型，設(shè)置模型評(píng)估器為ClusteringEvaluator，訓(xùn)練模型，并得到聚類(lèi)結(jié)果。完整實(shí)現(xiàn)如REF_Ref99401155\h代碼109所示，返回結(jié)果如REF_Ref99401695\h圖1018、REF_Ref119511141\h圖1019、REF_Ref99401711\h圖1020所示。代碼STYLEREF1\s10SEQ代碼\*ARABIC\s19K-means聚類(lèi)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")

/**

*修改需要進(jìn)行模型計(jì)算的特征字段為double類(lèi)型

*/

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

.selectExpr("cast(ageasdouble)age",

"cast(incomeasdouble)income",

"cast(spendingasdouble)spending",

"gender")

//獲取數(shù)據(jù)列名

valcolumns=data.columns

/**

*構(gòu)建模型，包含StringIndex（將gender性別字符數(shù)據(jù)轉(zhuǎn)換為Int類(lèi)型）、VectorAssembler（整合特征字段為Vector）

*和Kmeans模型。將詞三個(gè)模型放入到管道中

*/

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))

/**

*構(gòu)建Kmeans模型參數(shù)網(wǎng)格，設(shè)置最大迭代次數(shù)，隨機(jī)種子和k值

*使用訓(xùn)練集切分驗(yàn)證（默認(rèn)75%），獲取最佳參數(shù)模型，設(shè)置模型評(píng)估器為ClusteringEvaluator

*訓(xùn)練模型，并得到聚類(lèi)結(jié)果

*/

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