大數(shù)據(jù)分析及應(yīng)用項目教程(Spark SQL)（微課版） 課件 第4章 房產(chǎn)大數(shù)據(jù)分析與探索

第4章房產(chǎn)大數(shù)據(jù)分析與探索某房產(chǎn)公司銷售人員業(yè)績分析第一部分?jǐn)?shù)據(jù)集處理數(shù)據(jù)操作分析情境導(dǎo)入

員工業(yè)績考核分析是企業(yè)工作總結(jié)的重要組成部分。從業(yè)績分析中員工可以清楚的知道自己的工作應(yīng)該達(dá)到何種標(biāo)準(zhǔn)，發(fā)現(xiàn)自己的工作長處和不足，激發(fā)工作的積極性。管理者可以清楚的了解員工的工作情況，通過分析不足，幫助員工改進(jìn)、提高業(yè)績，促進(jìn)企業(yè)的發(fā)展進(jìn)步。假設(shè)現(xiàn)有某房產(chǎn)公司一個部門員工的某年銷售業(yè)績數(shù)據(jù)，請幫助部門經(jīng)理對其進(jìn)行統(tǒng)計分析，獲取相關(guān)信息。學(xué)習(xí)目標(biāo)和要求1、掌握將數(shù)據(jù)構(gòu)建為RDD的方法。2、能使用RDD的各種轉(zhuǎn)換和行動操作對數(shù)據(jù)集進(jìn)行處理分析。3、會對處理完成的數(shù)據(jù)進(jìn)行存儲操作。數(shù)據(jù)集處理

在Spark中最核心最基礎(chǔ)的概念是彈性分布式數(shù)據(jù)集（ResilientDistributedDatasets，RDD）。它是一種分布式的內(nèi)存抽象，可以基于任何數(shù)據(jù)結(jié)構(gòu)進(jìn)行創(chuàng)建。

創(chuàng)建RDD的方法有兩種，一種是基于內(nèi)存（集合）創(chuàng)建RDD，另一種是從外部數(shù)據(jù)集創(chuàng)建。1、基于內(nèi)存（集合）創(chuàng)建RDD。

從集合中創(chuàng)建RDD，主要提供了兩個方法：使用SparkContext類中的parallelize和makeRDD兩個方法。數(shù)據(jù)集處理（1）parallelize方法創(chuàng)建RDD

此方法有兩個參數(shù)可以輸入，第一個參數(shù)用來接收一個集合。第二個參數(shù)是可選的，用于指定創(chuàng)建的RDD的分區(qū)數(shù)。（2）makeRDD方法創(chuàng)建RDDmakeRDD方法和parallelize方法類似，但它可以指定每個分區(qū)的首選位置。數(shù)據(jù)集處理2、從外部存儲數(shù)據(jù)構(gòu)造RDD數(shù)據(jù)集處理

讀取一個放在文件系統(tǒng)中的數(shù)據(jù)進(jìn)行創(chuàng)建RDD。從本地文件創(chuàng)建，主要用于測試;使用HDFS文件獲取外部數(shù)據(jù)集創(chuàng)建,實踐操作中最常用。通過調(diào)用SparkContext的textFile方法讀取數(shù)據(jù)集。調(diào)用命令如下：textFile("/my/directory")textFile("/my/directory/*.txt")textFile("/my/directory/*.gz")（1）加載HDFS上存儲的CSV文件構(gòu)造一個RDD數(shù)據(jù)集處理

從HDFS中的數(shù)據(jù)創(chuàng)建RDD，首先需要定義文件在HDFS上存儲的路徑，然后根據(jù)路徑使用sc.textFile()方法進(jìn)行創(chuàng)建。//定義HDFS上test.csv文件的存儲路徑valHDFSfile="/Chapter4/test.csv"

//使用textFile方法，加載文件并構(gòu)造RDDvalHDFSrdd=sc.textFile(HDFSfile)

//查看返回值HDFSrdd.collect()（2）加載本地Linux中的文件構(gòu)造一個RDD數(shù)據(jù)集處理

讀取本地Linux中的文件構(gòu)造RDD也是通過sc.textFile("path")的方法，在path路徑前面加上“file:///”表示從本地文件系統(tǒng)讀取數(shù)據(jù)。3、操作練習(xí)——構(gòu)建員工業(yè)績RDD數(shù)據(jù)集處理（1）上傳數(shù)據(jù)集到HDFS文件系統(tǒng)中。（2）從HDFS文件系統(tǒng)中讀取文件并創(chuàng)建RDD。數(shù)據(jù)操作分析1、RDD支持的兩種操作。（1）轉(zhuǎn)換操作：就是對RDD中的數(shù)據(jù)進(jìn)行各種轉(zhuǎn)換。方法名稱作用map將RDD中的每一個數(shù)據(jù)元素通過func函數(shù)轉(zhuǎn)換，返回新的RDD。flatmap首先將map方法應(yīng)用于RDD的所有元素，然后將結(jié)果扁平化拆分，返回一個新的RDD。sortby通過指定條件對RDD中的元素進(jìn)行排序。filter通過指定條件對RDD中的元素進(jìn)行過濾。distinct對RDD中的所有元素去重，返回一個去重后的RDD。union將兩個RDD進(jìn)行合并，返回合并后的數(shù)據(jù)集。keys返回PairRDD中的“鍵”形成的新的RDD。values返回PairRDD中的“值”形成的新的RDD。reducebykey對“鍵”相同的值使用指定的函數(shù)進(jìn)行聚合操作。groupbykey對“鍵”相同的值根據(jù)指定條件進(jìn)行分組。sortbykey根據(jù)“鍵”對RDD內(nèi)部的元素進(jìn)行排序。join根據(jù)“鍵”對兩個RDD進(jìn)行連接。數(shù)據(jù)操作分析（2）行動操作：RDD的行動操作則是向驅(qū)動器程序返回結(jié)果或者把結(jié)果寫入外部系統(tǒng)的操作，會觸發(fā)實際的計算方法名稱作用count返回RDD中元素的個數(shù)。take返回RDD中前n個元素值。first返回RDD中第一個元素值。collect返回RDD中所有元素的列表。top返回RDD中排名前n的元素值。數(shù)據(jù)操作分析2、統(tǒng)計部門人員人數(shù)。count()行動操作，返回的是RDD內(nèi)元素的個數(shù)。舉例：利用序列Seq(1,2,3,4,5,6)創(chuàng)建的rdd中有6個數(shù)字元素操作練習(xí)：統(tǒng)計部門人員人數(shù)數(shù)據(jù)操作分析3、分別統(tǒng)計上/下半年業(yè)績排名，取前三位及最后三位利用map、flatmap、sortby、take，first、collect等方法對數(shù)據(jù)進(jìn)行操作實現(xiàn)。（1）map轉(zhuǎn)換操作map轉(zhuǎn)換操作是最常用的轉(zhuǎn)換算子，對RDD中的每個元素都執(zhí)行一個指定的函數(shù)來產(chǎn)生一個新的RDD。舉例：數(shù)據(jù)操作分析（2）flatmap轉(zhuǎn)換操作

此方法首先將map函數(shù)應(yīng)用于RDD的所有元素，然后將返回的結(jié)果平坦化。舉例：對map轉(zhuǎn)換操作中的str1進(jìn)行flatmap轉(zhuǎn)換操作。

對比map和flatMap操作的輸出結(jié)果圖，可以很容易的看出兩者的區(qū)別，因此flatMap操作通常用來切分單詞。數(shù)據(jù)操作分析（3）sortby轉(zhuǎn)換操作此方法是對RDD進(jìn)行排序，有3個參數(shù)可以輸入。 sortby(參數(shù)一,參數(shù)二,參數(shù)三)參數(shù)一是要進(jìn)行排序的對象值；參數(shù)二是排序方式，默認(rèn)是正序排序，使用false參數(shù)就是倒序排序；參數(shù)三是分區(qū)個數(shù)。數(shù)據(jù)操作分析//加載一個RDD，命名為testvaltest=sc.parallelize(Array(("dog",3),("cat",1),("monkey",2),("pig",3),("bird",2)))//根據(jù)test中的第二位元素進(jìn)行降序排序valtest_desc=test.sortBy(_._2,false)//查看結(jié)果test_desc.collect//根據(jù)test中的第二位元素進(jìn)行升序排序valtest_asc=test.sortBy(_._2)//查看結(jié)果test_asc.collectsortby舉例：數(shù)據(jù)操作分析（4）take，first、collect行動操作take行動操作返回RDD的前n個元素值。first行動操作返回的是RDD中的第一個元素值。collect行動操作是以數(shù)組的形式將RDD中所有的元素返回。數(shù)據(jù)操作分析（5）進(jìn)行分析，實現(xiàn)任務(wù)。1）按照前面的問題分析思路，首先對上半年和下半年的業(yè)績數(shù)據(jù)集firstrdd、secondrdd進(jìn)行map轉(zhuǎn)換，將每行數(shù)據(jù)分割為4列。2）利用sortBy方法對map_firstrdd和map_secondrdd數(shù)據(jù)集中的業(yè)績列分別進(jìn)行降序和升序排序，取出排名前三和后三的信息。對上半年業(yè)績列進(jìn)行降序和升序排序： valsort_firstrdd=map_firstrdd.sortBy(x=>x._4,false) sort_firstrdd.take(3) valsort_firstrdd=map_firstrdd.sortBy(x=>x._4) sort_firstrdd.take(3)同理對下半年業(yè)績列進(jìn)行降序和升序排序。數(shù)據(jù)操作分析4、統(tǒng)計上/下半年業(yè)績超過5000萬的人員。

要獲取這個問題的數(shù)據(jù)，首先需要分別過濾出上下半年業(yè)績超過5000萬的人員。但是有些人員只有上半年或者下半年符合要求，有些人員是上下半年都符合要求，因此還需要對過濾出來的人員信息進(jìn)行去重?？梢岳胒ilter、distinct、union等方法對數(shù)據(jù)進(jìn)行操作。（1）filter轉(zhuǎn)換操作此方法返回滿足指定過濾條件的元素，不滿足條件的元素被忽略。//讀取一個數(shù)字列表創(chuàng)建RDDvalrdd=sc.parallelize(List(1,3,6,2,8,8,5,6,7))//過濾出大于5的數(shù)字rdd.filter(x=>x>5).collect數(shù)據(jù)操作分析（2）distinct轉(zhuǎn)換操作

此方法是對RDD中的數(shù)據(jù)去重，把完全相同的元素去除。對前述filter轉(zhuǎn)換操作中的rdd數(shù)據(jù)進(jìn)行操作。由輸出結(jié)果可見，原先列表中重復(fù)的數(shù)字6和8被去重了。數(shù)據(jù)操作分析（3）union轉(zhuǎn)換操作

此方法可以將兩個RDD進(jìn)行合并，返回兩個RDD的并集，并且不去重。但是要求兩個RDD中每個元素中的值的個數(shù)及數(shù)據(jù)類型保持一致。//定義加載兩個RDD，命名為test1和test2valtest1=sc.parallelize(Array(("dog",3),("cat",1),("monkey",2)))valtest2=sc.parallelize(Array(("monkey",2),("pig",3),("bird",2)))//執(zhí)行union操作合并兩個RDDtest1.union(test2).collect數(shù)據(jù)操作分析（4）進(jìn)行分析，實現(xiàn)任務(wù)。1）利用filter方法過濾出業(yè)績大于5000萬的人員，最終返回人員編號信息。2）利用union方法，將上述過濾后的數(shù)據(jù)集filter_firstrdd和filter_secondrdd進(jìn)行合并，并通過distinct方法去除重復(fù)的人員編號信息。數(shù)據(jù)操作分析5、統(tǒng)計此部門當(dāng)年的房屋銷售套數(shù)。

要獲取這個問題的數(shù)據(jù)，只需取出數(shù)據(jù)表的第三列“房屋銷售套數(shù)”數(shù)據(jù)，然后利用求和函數(shù)進(jìn)行求和。可以利用RDD的相關(guān)描述性統(tǒng)計函數(shù)實現(xiàn)。（1）min()、max()函數(shù)min()方法返回RDD中的最小值，max()方法返回RDD中的最大值。數(shù)據(jù)操作分析（2）mean()、sum()函數(shù)mean()函數(shù)返回RDD中的平均值。sum()函數(shù)返回RDD中的總和。舉例：求RDD數(shù)據(jù)集的平均值和總和。數(shù)據(jù)操作分析（3）variance()、stdev()函數(shù)variance()計算RDD中所有元素的總體方差。stdev()計算RDD的標(biāo)準(zhǔn)差。舉例：求RDD數(shù)據(jù)集的總體方差和樣本方差。數(shù)據(jù)操作分析（4）進(jìn)行分析，實現(xiàn)任務(wù)。

將經(jīng)過map操作處理好的上/下半年業(yè)績數(shù)據(jù)集map_firstrdd和map_secondrdd進(jìn)行合并，取出第三列數(shù)據(jù)，利用sum函數(shù)計算總數(shù)。數(shù)據(jù)操作分析6、查看全年總業(yè)績最高人員。

此任務(wù)需要將每個人上、下半年的業(yè)績進(jìn)行相加，然后取最高者。這個過程需要對人員編號相同的業(yè)績數(shù)據(jù)進(jìn)行識別并相加。在RDD中可以通過鍵值對（Key-Value）的操作完成。Spark為包含鍵值對類型的RDD提供了一些專有的操作，這些RDD被稱為Pair

RDD。數(shù)據(jù)操作分析（1）創(chuàng)建PairRDD

創(chuàng)建PairRDD的方法有很多，當(dāng)需要將一個普通RDD轉(zhuǎn)換為PairRDD時，可以使用map方法來實現(xiàn)。//定義一個RDDvalrdd=sc.parallelize(List("dog","cat","monkey","pig"))//創(chuàng)建鍵值對valpairrdd=rdd.map(word=>(word,1))（2）keys()、values()轉(zhuǎn)換操作keys()、values()操作返回的類型是RDD。//取pairrdd中的鍵pairrdd.keys.collect//取pairrdd中的值pairrdd.values.collect數(shù)據(jù)操作分析（3）reducebykey()轉(zhuǎn)換操作

此方法應(yīng)用于鍵值對數(shù)據(jù)集操作，對Key相同的Value使用指定的函數(shù)進(jìn)行聚合操作，返回一個鍵值對的數(shù)據(jù)集。reduceByKey((x,y)=>x+y)，對PairRDD中鍵相同的值進(jìn)行相加操作。數(shù)據(jù)操作分析（4）groupbykey()轉(zhuǎn)換操作

此方法會對相同鍵的值進(jìn)行分組，形成二元元組，第一個字段為相同的鍵，第二個字段為具備相同鍵的值的集合。對前一個任務(wù)中的map_test數(shù)據(jù)集進(jìn)行操作。 valgroup_test=map_test.groupByKey() group_test.collect數(shù)據(jù)操作分析（5）sortByKey()轉(zhuǎn)換操作此方法返回一個根據(jù)“鍵”進(jìn)行排序的RDD。舉例：數(shù)據(jù)操作分析（6）進(jìn)行分析，實現(xiàn)任務(wù)1）將上、下半年的的業(yè)績合并到同一個RDD中。valtotal=map_firstrdd.union(map_secondrdd)2）將數(shù)據(jù)轉(zhuǎn)換成（員工編號，業(yè)績）鍵值對，利用reduceByKey方法對相同員工編號對應(yīng)的業(yè)績值進(jìn)行相加。數(shù)據(jù)操作分析3）對業(yè)績總和進(jìn)行排序，獲得最高業(yè)績?nèi)藛T信息。//利用sortBy方法對指定列進(jìn)行排序 yeji.sortBy(x=>x._2,false) yeji.sortBy(x=>x._2,false).take(1)數(shù)據(jù)操作分析7、存儲以上統(tǒng)計分析信息

數(shù)據(jù)存儲的方式有多種，比如將數(shù)據(jù)存儲為JSON文件、CSV文件、文本文件等。文本文件的存儲可以直接調(diào)用saveAsTextFile(path)進(jìn)行存儲。

對上、下半年的業(yè)績以及全年總業(yè)績數(shù)據(jù)進(jìn)行join連接，再將結(jié)果數(shù)據(jù)以文本文件的形式存儲到HDFS中。操作練習(xí)：某城市近年房產(chǎn)銷售狀況分析第二部分?jǐn)?shù)據(jù)準(zhǔn)備數(shù)據(jù)探索與分析情境導(dǎo)入

房子是一個價格昂貴的“商品”，但也是每個人生活必不可少的一件“商品”。房子的價格、所處的位置、裝修情況等等因素都是購房者關(guān)心的主要問題。了解房產(chǎn)市場的變化規(guī)律以及各個因素對價格的影響，對于消費者來說是購房之前首先要做的功課。

現(xiàn)有一份來自某網(wǎng)站的某城市近年房產(chǎn)銷售數(shù)據(jù)集house-price.csv，包含成交時間、價格、裝修情況、區(qū)域等信息。學(xué)習(xí)目標(biāo)和要求

能靈活綜合的應(yīng)用各種RDD操作和各個算子對數(shù)據(jù)進(jìn)行分析，解決以下問題： 1、每年房產(chǎn)銷售量趨勢如何？ 2、此城市各區(qū)域房產(chǎn)銷量如何？均價如何？ 3、在2018年1月1日的銷量有多少？ 4、查詢2018年1月1日到1月31日之間，滿五年房產(chǎn)的銷售比例。 5、哪種裝修類型的房子銷量最高？ 6、所售賣的房子中，電梯有無的比例，地鐵有無情況？數(shù)據(jù)準(zhǔn)備

此房產(chǎn)銷售數(shù)據(jù)集是一個在系統(tǒng)外部的csv文件，因此需要將其上傳至文件系統(tǒng)中，加載為RDD后再做處理分析。1、利用MobaXterm工具，將house-price.csv文件上傳到本地Linux的/root/data/Chapter4文件夾內(nèi)。2、啟動Hadoop和Spark集群，將/root/data/Chapter4目錄下的house-price.csv文件上傳到HDFS文件系統(tǒng)的/Chapter4目錄中。3、啟動spark-shell，加載數(shù)據(jù)到RDD。數(shù)據(jù)探索與分析

在進(jìn)行數(shù)據(jù)探索分析之前，還需要將前面加載到RDD中的數(shù)據(jù)進(jìn)行轉(zhuǎn)換，把每一條數(shù)據(jù)根據(jù)字段含義進(jìn)行分割，以便后續(xù)數(shù)據(jù)處理分析。1、使用map方法轉(zhuǎn)換RDD，以“,”將每一行數(shù)據(jù)分割。數(shù)據(jù)探索與分析2、對戶人數(shù)的基本情況進(jìn)行統(tǒng)計，獲得記錄條目的總數(shù)，以及在這些記錄中包括多少個用戶ID。//查看數(shù)據(jù)總條數(shù)m_houseprice.count//查看這些記錄中的用戶數(shù)m_houseprice.map(_(1)).distinct.count數(shù)據(jù)探索與分析3、了解房產(chǎn)銷量趨勢，統(tǒng)計每年房產(chǎn)銷售量情況。

要獲得每年的房產(chǎn)銷售量情況，需要先從日期中提取年份數(shù)據(jù)；然后利用reduceByKey方法按年進(jìn)行統(tǒng)計匯總；最后利用sortBy方法對統(tǒng)計匯總的數(shù)據(jù)進(jìn)行排序，獲得每年銷量從高到低的排序情況。數(shù)據(jù)探索與分析4、分析此數(shù)據(jù)集中購買房屋所屬的區(qū)域是哪些？各個區(qū)域的房產(chǎn)銷售情況如何？各個區(qū)域的房屋均價