第3章 Spark編程基礎(chǔ)

Spark編程基礎(chǔ)2021年，某公司為了提高員工工作的積極性，將對公司員工進行一次調(diào)薪，需要根據(jù)員工在2020年的薪資情況及在職表現(xiàn)重新調(diào)整薪資，對于愛崗敬業(yè)的公司員工，公司擬根據(jù)其業(yè)績分析情況予以不同程度漲薪。公司有員工2020年上半年薪資文件（Employee_salary_first_half.csv）和下半年薪資文件（Employee_salary_second_half.csv），兩份文件的數(shù)據(jù)格式和數(shù)據(jù)字段均相同，以員工2020年上半年的薪資文件Employee_salary_first_half.csv為例，文件共有10個數(shù)據(jù)字段。任務(wù)背景字段名稱說明

字段名稱說明EmpID員工IDGROSS總薪資Name姓名Net_Pay實際薪資Gender性別Deduction薪資扣除部分Date_of_Birth出生日期Designation職位Age年齡Department部門為了保證較高的數(shù)據(jù)處理效率，將使用Spark統(tǒng)計每一位員工2020年的薪資情況。本章將介紹SparkRDD的創(chuàng)建方法、RDD和鍵值對RDD的轉(zhuǎn)換操作和行動操作的基礎(chǔ)使用，并通過Spark編程實現(xiàn)員工薪資數(shù)據(jù)的統(tǒng)計分析。任務(wù)背景1查詢上半年實際薪資排名前3的員工信息目錄讀取員工薪資數(shù)據(jù)創(chuàng)建RDD2查詢上半年或下半年實際薪資大于20萬元的員工姓名3RDD是一個容錯的、只讀的、可進行并行操作的數(shù)據(jù)結(jié)構(gòu)，是一個分布在集群各個節(jié)點中的存放元素的集合。RDD的創(chuàng)建有3種不同的方法。第一種是將程序中已存在的Seq集合（如集合、列表、數(shù)組）轉(zhuǎn)換成RDD。第二種是對已有RDD進行轉(zhuǎn)換得到新的RDD，這兩種方法都是通過內(nèi)存中已有的集合創(chuàng)建RDD的。第三種是直接讀取外部存儲系統(tǒng)的數(shù)據(jù)創(chuàng)建RDD。為方便后續(xù)的員工薪資分析，本節(jié)的任務(wù)是讀取員工上半年和下半年的薪資數(shù)據(jù)創(chuàng)建RDD。任務(wù)描述parallelize()方法有兩個輸入?yún)?shù)，說明如下。要轉(zhuǎn)化的集合，必須是Seq集合。Seq表示序列，指的是一類具有一定長度的、可迭代訪問的對象，其中每個數(shù)據(jù)元素均帶有一個從0開始的、固定的索引。分區(qū)數(shù)。若不設(shè)分區(qū)數(shù)，則RDD的分區(qū)數(shù)默認為該程序分配到的資源的CPU核心數(shù)。從內(nèi)存中讀取數(shù)據(jù)創(chuàng)建RDD1.parallelize()makeRDD()方法有兩種使用方式。第一種方式的使用與parallelize()方法一致；第二種方式是通過接收一個是Seq[(T,Seq[String])]參數(shù)類型創(chuàng)建RDD。第二種方式生成的RDD中保存的是T的值，Seq[String]部分的數(shù)據(jù)會按照Seq[(T,Seq[String])]的順序存放到各個分區(qū)中，一個Seq[String]對應(yīng)存放至一個分區(qū)，并為數(shù)據(jù)提供位置信息，通過preferredLocations()方法可以根據(jù)位置信息查看每一個分區(qū)的值。調(diào)用makeRDD()時不可以直接指定RDD的分區(qū)個數(shù)，分區(qū)的個數(shù)與Seq[String]參數(shù)的個數(shù)是保持一致的。從內(nèi)存中讀取數(shù)據(jù)創(chuàng)建RDD2．makeRDD()從外部存儲系統(tǒng)中讀取數(shù)據(jù)創(chuàng)建RDD是指直接讀取存放在文件系統(tǒng)中的數(shù)據(jù)文件創(chuàng)建RDD。從內(nèi)存中讀取數(shù)據(jù)創(chuàng)建RDD的方法常用于測試，從外部存儲系統(tǒng)中讀取數(shù)據(jù)創(chuàng)建RDD才是用于實踐操作的常用方法。從外部存儲系統(tǒng)中讀取數(shù)據(jù)創(chuàng)建RDD可以有很多種數(shù)據(jù)來源，可通過SparkContext對象的textFile()方法讀取數(shù)據(jù)集，該方法支持多種類型的數(shù)據(jù)集，如目錄、文本文件、壓縮文件和通配符匹配的文件等，并且允許設(shè)定分區(qū)個數(shù)。從外部存儲系統(tǒng)中讀取數(shù)據(jù)創(chuàng)建RDD分別讀取HDFS文件和Linux本地文件的數(shù)據(jù)并創(chuàng)建RDD，具體操作如下。通過HDFS文件創(chuàng)建RDD直接通過textFile()方法讀取HDFS文件的位置即可。通過Linux本地文件創(chuàng)建RDD本地文件的讀取也是通過sc.textFile("路徑")的方法實現(xiàn)的，在路徑前面加上“file://”表示從Linux本地文件系統(tǒng)讀取。在IntelliJIDEA開發(fā)環(huán)境中可以直接讀取本地文件；但在spark-shell中，要求在所有節(jié)點的相同位置保存該文件才可以讀取它.從外部存儲系統(tǒng)中讀取數(shù)據(jù)創(chuàng)建RDD讀取員工上、下半年薪資數(shù)據(jù)，并創(chuàng)建RDD。由于數(shù)據(jù)比較多，因此適合通過讀取HDFS上的數(shù)據(jù)創(chuàng)建。首先需要將數(shù)據(jù)上傳至HDFS的/user/root目錄下。在spark-shell中讀取HDFS上的員工上、下半年薪資數(shù)據(jù)，創(chuàng)建RDD。任務(wù)實現(xiàn)1查詢上半年實際薪資排名前3的員工信息目錄讀取員工薪資數(shù)據(jù)創(chuàng)建RDD2查詢上半年或下半年實際薪資大于20萬元的員工姓名3SparkRDD提供了豐富的操作方法用于操作分布式的數(shù)據(jù)集合，包括轉(zhuǎn)換操作和行動操作兩部分。轉(zhuǎn)換操作可以將一個RDD轉(zhuǎn)換為一個新的RDD，但是轉(zhuǎn)換操作是懶操作，不會立刻執(zhí)行計算；行動操作是用于觸發(fā)轉(zhuǎn)換操作的操作，這時才會真正開始進行計算。本節(jié)的任務(wù)如下。使用RDD的基本操作完成對員工上半年實際薪資的排名。找出薪資排名前3的員工信息。任務(wù)描述map()方法是一種基礎(chǔ)的RDD轉(zhuǎn)換操作，可以對RDD中的每一個數(shù)據(jù)元素通過某種函數(shù)進行轉(zhuǎn)換并返回新的RDD。map()方法是轉(zhuǎn)換操作，不會立即進行計算。轉(zhuǎn)換操作是創(chuàng)建RDD的第二種方法，通過轉(zhuǎn)換已有RDD生成新的RDD。因為RDD是一個不可變的集合，所以如果對RDD數(shù)據(jù)進行了某種轉(zhuǎn)換，那么會生成一個新的RDD。使用map()方法轉(zhuǎn)換數(shù)據(jù)sortBy()方法用于對標準RDD進行排序，有3個可輸入?yún)?shù)，說明如下。第1個參數(shù)是一個函數(shù)f:(T)=>K，左邊是要被排序?qū)ο笾械拿恳粋€元素，右邊返回的值是元素中要進行排序的值。第2個參數(shù)是ascending，決定排序后RDD中的元素是升序的還是降序的，默認是true，即升序排序，如果需要降序排序那么需要將參數(shù)的值設(shè)置為false。第3個參數(shù)是numPartitions，決定排序后的RDD的分區(qū)個數(shù)，默認排序后的分區(qū)個數(shù)和排序之前的分區(qū)個數(shù)相等，即this.partitions.size。第一個參數(shù)是必須輸入的，而后面的兩個參數(shù)可以不輸入。使用sortBy()方法進行排序collect()方法是一種行動操作，可以將RDD中所有元素轉(zhuǎn)換成數(shù)組并返回到Driver端，適用于返回處理后的少量數(shù)據(jù)。因為需要從集群各個節(jié)點收集數(shù)據(jù)到本地，經(jīng)過網(wǎng)絡(luò)傳輸，并且加載到Driver內(nèi)存中，所以如果數(shù)據(jù)量比較大，會給網(wǎng)絡(luò)傳輸造成很大的壓力。因此，數(shù)據(jù)量較大時，盡量不使用collect()方法，否則可能導(dǎo)致Driver端出現(xiàn)內(nèi)存溢出問題。使用collect()方法查詢數(shù)據(jù)collect()方法有以下兩種操作方式。collect：直接調(diào)用collect返回該RDD中的所有元素，返回類型是一個Array[T]數(shù)組。collect[U:ClassTag](f:PartialFunction[T,U])：RDD[U]。這種方式需要提供一個標準的偏函數(shù)，將元素保存至一個RDD中。首先定義一個函數(shù)one，用于將collect方法得到的數(shù)組中數(shù)值為1的值替換為“one”，將其他值替換為“other”。使用collect()方法查詢數(shù)據(jù)flatMap()方法將函數(shù)參數(shù)應(yīng)用于RDD之中的每一個元素，將返回的迭代器（如數(shù)組、列表等）中的所有元素構(gòu)成新的RDD。使用flatMap()方法時先進行map（映射）再進行flat（扁平化）操作，數(shù)據(jù)會先經(jīng)過跟map一樣的操作，為每一條輸入返回一個迭代器（可迭代的數(shù)據(jù)類型），然后將所得到的不同級別的迭代器中的元素全部當成同級別的元素，返回一個元素級別全部相同的RDD。這個轉(zhuǎn)換操作通常用來切分單詞。使用flatMap()方法轉(zhuǎn)換數(shù)據(jù)take(N)方法用于獲取RDD的前N個元素，返回數(shù)據(jù)為數(shù)組。take()與collect()方法的原理相似，collect()方法用于獲取全部數(shù)據(jù)，take()方法獲取指定個數(shù)的數(shù)據(jù)。獲取RDD的前5個元素使用take()方法查詢某幾個值查詢上半年實際薪資排名前3的員工信息，需要對上半年的實際薪資進行排序，而創(chuàng)建RDD時，textFile()方法是將每一行數(shù)據(jù)作為一條記錄存儲的，所以在排序前需要先對數(shù)據(jù)進行轉(zhuǎn)換，實現(xiàn)步驟如下。讀取CSV文件，將第一行字段名稱刪除。將數(shù)據(jù)按分隔符“,”分隔，取出第2列員工姓名和第7列實際薪資數(shù)據(jù)，并將實際薪資數(shù)據(jù)轉(zhuǎn)換成Int類型數(shù)據(jù)。通過sortBy()方法根據(jù)實際薪資進行降序排列。通過take()方法獲取上半年實際薪資排名前3的員工信息。任務(wù)實現(xiàn)1查詢上半年實際薪資排名前3的員工信息目錄讀取員工薪資數(shù)據(jù)創(chuàng)建RDD2查詢上半年或下半年實際薪資大于20萬元的員工姓名3Spark的轉(zhuǎn)換操作和行動操作，除了可以針對一個RDD進行操作，也可以進行RDD與RDD之間的操作。本節(jié)的任務(wù)如下。查詢上半年或下半年實際薪資大于20萬元的員工姓名。將最終結(jié)果合并為一個RDD。任務(wù)描述union()方法是一種轉(zhuǎn)換操作，用于將兩個RDD合并成一個，不進行去重操作，而且兩個RDD中每個元素中的值的個數(shù)、數(shù)據(jù)類型需要保持一致。使用union()方法合并兩個RDD使用union()方法合并多個RDDfilter()方法是一種轉(zhuǎn)換操作，用于過濾RDD中的元素。filter()方法需要一個參數(shù)，這個參數(shù)是一個用于過濾的函數(shù)，該函數(shù)的返回值為Boolean類型。filter()方法將返回值為true的元素保留，將返回值為false的元素過濾掉，最后返回一個存儲符合過濾條件的所有元素的新RDD。創(chuàng)建一個RDD，并且過濾掉每個元組第二個值小于等于1的元素。使用filter()方法進行過濾distinct()方法是一種轉(zhuǎn)換操作，用于RDD的數(shù)據(jù)去重，去除兩個完全相同的元素，沒有參數(shù)。創(chuàng)建一個帶有重復(fù)數(shù)據(jù)的RDD，并使用distinct()方法去重。使用distinct()方法進行去重Spark中的集合操作常用方法（轉(zhuǎn)換操作）使用簡單的集合操作方法描述union()參數(shù)是RDD，合并兩個RDD的所有元素intersection()參數(shù)是RDD，求出兩個RDD的共同元素subtract()參數(shù)是RDD，將原RDD里和參數(shù)RDD里相同的元素去掉cartesian()參數(shù)是RDD，求兩個RDD的笛卡兒積intersection()方法用于求出兩個RDD的共同元素，即找出兩個RDD的交集，參數(shù)是另一個RDD，先后順序與結(jié)果無關(guān)。創(chuàng)建兩個RDD，其中有相同的元素，通過intersection()方法求出兩個RDD的交集。使用簡單的集合操作intersection()方法subtract()方法用于將前一個RDD中在后一個RDD出現(xiàn)的元素刪除，可以認為是求補集的操作，返回值為前一個RDD去除與后一個RDD相同元素后的剩余值所組成的新的RDD。兩個RDD的順序會影響結(jié)果。創(chuàng)建兩個RDD，分別為rdd1和rdd2，包含相同元素和不同元素，通過subtract()方法求rdd1和rdd2彼此的補集。使用簡單的集合操作subtract()方法cartesian()方法可將兩個集合的元素兩兩組合成一組，即求笛卡兒積。創(chuàng)建兩個RDD，分別有4個元素，通過cartesian()方法求兩個RDD的笛卡兒積。使用簡單的集合操作cartesian()方法輸出上半年或下半年實際薪資大于20萬元的員工姓名。首先需要過濾出兩個RDD中實際薪資大于20萬元的員工姓名。再將兩個RDD得到的員工姓名合并到一個RDD中，對員工姓名進行去重。即可得到上半年或下半年實際薪資大于20萬元的員工姓名。任務(wù)實現(xiàn)4查詢每位員工2020年的月均實際薪資目錄輸出每位員工2020年的總實際薪資5存儲匯總后的員工薪資為文本文件6鍵值對RDD存儲二元組，二元組分為鍵和值，RDD的基本轉(zhuǎn)換操作對于鍵值對RDD也同樣適用。因為鍵值對RDD中包含的是二元組，所以需要傳遞的函數(shù)會由原來的操作單個元素的函數(shù)改為操作二元組的函數(shù)。本節(jié)的任務(wù)如下。計算每位員工2020年的總實際薪資，要求對上、下半年員工的實際薪資進行相加。任務(wù)描述Spark的大部分RDD操作都支持所有種類的單值RDD，但是有少部分特殊的操作只能作用于鍵值對類型的RDD。顧名思義，鍵值對RDD由一組組的鍵值對組成，這些RDD被稱為PairRDD。PairRDD提供了并行操作各個鍵或跨節(jié)點重新進行數(shù)據(jù)分組的操作接口。例如，PairRDD提供了reduceByKey()方法，可以分別規(guī)約每個鍵對應(yīng)的數(shù)據(jù)，還有join()方法，可以把兩個RDD中鍵相同的元素組合在一起，合并為一個RDD。了解鍵值對RDD有很多種創(chuàng)建鍵值對RDD的方式，很多存儲鍵值對的數(shù)據(jù)格式會在讀取時直接返回由其鍵值對組成的PairRDD。當需要將一個普通的RDD轉(zhuǎn)化為一個PairRDD時可以使用map函數(shù)來進行操作，傳遞的函數(shù)需要返回鍵值對。創(chuàng)建鍵值對RDD使用鍵值對RDD的keys和values方法鍵值對RDD，包含鍵和值兩個部分。Spark提供了兩種方法，分別獲取鍵值對RDD的鍵和值。keys方法返回一個僅包含鍵的RDD。values方法返回一個僅包含值的RDD。當數(shù)據(jù)集以鍵值對形式展現(xiàn)時，合并統(tǒng)計鍵相同的值是很常用的操作。reduceByKey()方法用于合并具有相同鍵的值，作用對象是鍵值對，并且只對每個鍵的值進行處理，當RDD中有多個鍵相同的鍵值對時，則會對每個鍵對應(yīng)的值進行處理。reduceByKey()方法需要接收一個輸入函數(shù)，鍵值對RDD相同鍵的值會根據(jù)函數(shù)進行合并并且創(chuàng)建一個新的RDD作為返回結(jié)果。使用鍵值對RDD的reduceByKey()方法在進行處理時，reduceByKey()方法將相同鍵的前兩個值傳給輸入函數(shù)，產(chǎn)生一個新的返回值，新產(chǎn)生的返回值與RDD中相同鍵的下一個值組成兩個元素，再傳給輸入函數(shù)，直到最后每個鍵只有一個對應(yīng)的值為止。reduceByKey()方法不是一種行動操作，而是一種轉(zhuǎn)換操作。使用鍵值對RDD的reduceByKey()方法groupByKey()方法用于對具有相同鍵的值進行分組，可以對同一組的數(shù)據(jù)進行計數(shù)、求和等操作。對于一個由類型K的鍵和類型V的值組成的RDD，通過groupByKey()方法得到的RDD類型是[K,Iterable[V]]。使用鍵值對RDD的groupByKey()方法在讀取上、下半年員工薪資數(shù)據(jù)并將其轉(zhuǎn)換為RDD的過程中，已經(jīng)將數(shù)據(jù)轉(zhuǎn)換成鍵值對RDD。統(tǒng)計每位員工2020年的總實際薪資，首先需要將數(shù)據(jù)合并到一個RDD中，通過相同的鍵對同一個員工的上半年實際薪資和下半年實際薪資進行累加，實現(xiàn)步驟如下。獲取上、下半年員工薪資數(shù)據(jù)并將其轉(zhuǎn)換為RDD，分別為split_first和split_second。使用union()方法將兩個RDD合并成一個新的RDD。通過reduceByKey()方法統(tǒng)計員工總實際薪資并輸出結(jié)果。任務(wù)實現(xiàn)4查詢每位員工2020年的月均實際薪資目錄輸出每位員工2020年的總實際薪資5存儲匯總后的員工薪資為文本文件6在Spark中，鍵值對RDD提供了很多基于多個RDD的鍵進行操作的方法。本節(jié)的任務(wù)是輸出每位員工2020年的每月平均實際薪資，需要先計算每位員工2020年的總實際薪資，再求出每位員工2020年的月均實際薪資。任務(wù)描述將有鍵的一組數(shù)據(jù)與另一組有鍵的數(shù)據(jù)根據(jù)鍵進行連接，是對鍵值對數(shù)據(jù)常用的操作之一。與合并不同，連接會對鍵相同的值進行合并，連接方式多種多樣，包含內(nèi)連接、右外連接、左外連接、全外連接，不同的連接方式需要使用不同的連接方法。連接方法如下表。使用join()方法連接兩個RDD連接方法描述join()對兩個RDD進行內(nèi)連接rightOuterJoin()對兩個RDD進行連接操作，確保第二個RDD的鍵必須存在（右外連接）leftOuterJoin()對兩個RDD進行連接操作，確保第一個RDD的鍵必須存在（左外連接）fullOuterJoin()對兩個RDD進行全外連接join()方法用于根據(jù)鍵對兩個RDD進行內(nèi)連接，將兩個RDD中鍵相同的數(shù)據(jù)的值存放在一個元組中，最后只返回兩個RDD中都存在的鍵的連接結(jié)果。例如，在兩個RDD中分別有鍵值對(K,V)和(K,W)，通過join()方法連接會返回(K,(V,W))。創(chuàng)建兩個RDD，含有相同鍵和不同的鍵，通過join()方法進行內(nèi)連接。使用join()方法連接兩個RDD（1）join()方法rightOuterJoin()方法用于根據(jù)鍵對兩個RDD進行右外連接，連接結(jié)果是右邊RDD的所有鍵的連接結(jié)果，不管這些鍵在左邊RDD中是否存在。在rightOuterJoin()方法中，如果在左邊RDD中有對應(yīng)的鍵，那么連接結(jié)果中值顯示為Some類型值；如果沒有，那么顯示為None值。使用join()方法連接兩個RDD（2）rightOuterJoin()方法leftOuterJoin()方法用于根據(jù)鍵對兩個RDD進行左外連接，與rightOuterJoin()方法相反，返回結(jié)果保留左邊RDD的所有鍵。使用join()方法連接兩個RDD（3）leftOuterJoin()方法fullOuterJoin()方法用于對兩個RDD進行全外連接，保留兩個RDD中所有鍵的連接結(jié)果。使用join()方法連接兩個RDD（4）fullOuterJoin()方法zip()方法用于將兩個RDD組合成鍵值對RDD，要求兩個RDD的分區(qū)數(shù)量以及元素數(shù)量相同，否則會拋出異常。將兩個RDD組合成Key/Value形式的RDD，這里要求兩個RDD的partition數(shù)量以及元素數(shù)量都相同，否則會拋出異常。使用zip()方法組合兩個RDDcombineByKey()方法是Spark中一個比較核心的高級方法，鍵值對的其他一些高級方法底層均是使用combineByKey()方法實現(xiàn)的，如groupByKey()方法、reduceByKey()方法等。combineByKey()方法用于將鍵相同的數(shù)據(jù)聚合，并且允許返回類型與輸入數(shù)據(jù)的類型不同的返回值。combineByKey()方法的使用方式如下。combineByKey(createCombiner,mergeValue,mergeCombiners,numPartitions=None)使用combineByKey()方法合并相同鍵的值combineByKey()方法接收3個重要的參數(shù)，具體說明如下。createCombiner:V=>C，V是鍵值對RDD中的值部分，將該值轉(zhuǎn)換為另一種類型的值C，C會作為每一個鍵的累加器的初始值。mergeValue:(C,V)=>C，該函數(shù)將元素V聚合到之前的元素C(createCombiner)上（這個操作在每個分區(qū)內(nèi)進行）。mergeCombiners:(C,C)=>C，該函數(shù)將兩個元素C進行合并（這個操作在不同分區(qū)間進行)。使用combineByKey()方法合并相同鍵的值由于合并操作會遍歷分區(qū)中所有的元素，因此每個元素（這里指的是鍵值對）的鍵只有兩種情況：以前沒出現(xiàn)過或以前出現(xiàn)過。對于這兩種情況，3個參數(shù)的執(zhí)行情況描述如下。如果以前沒出現(xiàn)過，則執(zhí)行的是createCombiner()方法，createCombiner()方法會在新遇到的鍵對應(yīng)的累加器中賦予初始值，否則執(zhí)行mergeValue()方法。對于已經(jīng)出現(xiàn)過的鍵，調(diào)用mergeValue()方法進行合并操作，對該鍵的累加器對應(yīng)的當前值（C）與新值（V）進行合并。由于每個分區(qū)都是獨立處理的，因此對于同一個鍵可以有多個累加器。如果有兩個或更多的分區(qū)都有對應(yīng)同一個鍵的累加器，就需要使用用戶提供的mergeCombiners()方法對各個分區(qū)的結(jié)果（全是C）進行合并。使用combineByKey()方法合并相同鍵的值lookup(key:K)方法作用于鍵值對RDD，返回指定鍵的所有值。使用lookup()方法查找指定鍵的值查詢每位員工2020年的月均實際薪資需要先篩選出上、下半年的員工薪資數(shù)據(jù)中的員工姓名和實際薪資兩個字段數(shù)據(jù)并創(chuàng)建RDD，然后將篩選后的兩個RDD合并，再根據(jù)員工姓名對實際薪資求和，最后查詢出2020年的每位員工的月均實際薪資，具體實現(xiàn)步驟如下。獲取兩個RDD，即split_first和split_second，使用union()方法合并兩個RDD。使用combineByKey()方法計算每位員工2020年的月均實際薪資。任務(wù)實現(xiàn)4查詢每位員工2020年的月均實際薪資目錄輸出每位員工2020年的總實際薪資5存儲匯總后的員工薪資為文本文件6在實際生產(chǎn)環(huán)境中，需要讀取的文本格式不僅包含普通的文本文件，還包含其他格式的文件，如JS