Spark shell 詞頻統計和統計PV心得
並以本人能夠接受的方式理解的，大家可以參考，如有問題請留言指正。
樣本數據
[hadoop@h201 ~]$ cat hh.txt
hello,world
hello,hadoop
hello,oracle
hadoop,oracle
hello,world
hello,hadoop
hello,oracle
hadoop,oracle
詞頻統計，及其按單詞數量倒序排序過程及其詳解
1.將文件加載成RDD
Scala> var file=sc.textFile(“hdfs://h201:9000/hh.txt”)
2.將每行按逗號拆分，結果裝載到一個數組中，每次提取一個單詞， _代表每次輸入內容的占位符
Scala> val h1=file.flatMap(_.split(
3. 將數組中的每個元素裝載到map方法中執行統一的處理任務,將輸入的每個單詞返回成k,v 鍵值對,reduceByKey()方法只對value只運行括號內的方法進行叠代計算_+_ 代表累加，返回的是k和進行過叠代計算的v 鍵值對
Scala> val h2=h1.map(x=>(x,1)).reduceByKey(_+_)
4. 再用第二個map接收上一步的k,v鍵值對進行交換位置輸出例如：
輸入的是（“hello”,5）變成（5,”hello”）
Scala> val h3=h2.map(_.2,_.1)
5. 將結果按key值排序
Scala> val h4=h4.sortByKey(false) false=倒序
6. 在使用map函數將拍好序的鍵值對進行交換例如：
(5,”hello”) (4,”hadoop”) 變成(“hello”,5)(“hadoop”,4)
Scala> val h5=h4.map(_.2,_.1)
7. 到此已經完成了詞頻統計並按照單詞數量的降序進行了排列已經完成下一步可以將結果輸出到文件夾中，註意是一個目錄
Scala> h5.saveAsTextFile("hdfs://h201:9000/output1")
上述所有操作拆分為了方便理解，可以將所有操作合成一條代碼：如下
Scala > val wc = file.flatMap(_.split(",")).map(x=>(x,1)).reduceByKey(_+_).map(x=>(x._2,x._1)).sortByKey(false).map(x=>(x._2,x._1)).saveAsTextFile(
flatMap() 與 map() 的區別
flatMap() 與 map() 都是對輸入的每行內容做同樣的操作但是產生的結果不相同；
例如樣本：
hello,world
hello,hadoop
hello,oracle
將文件導入成RDD =》var file=sc.textFile(“hdfs://xxx:9000/xx.txt”)
同樣是用split方法按逗號分隔
Var fm=file.flatMap(_.split(“,”)) 每行按逗號分隔後產生的結果解將每個單詞放在一個集合中，下面如果使用fm中的內容是每次只會導入一個單詞：
用java表示就是{‘hello’,’world’,’hello’,’hadoop’,’hello’,’oracle’} 相當於一維數組
Var m=file.map(_.split(“,”)) 每行按逗號分隔後產生的結果是將每行的變成一個字符串數組，再放到一個大的結果集中,下面如果使用m中的內容每次導入一個數組：
用java表示就是{{‘hello’,’world’},{‘hello’,’hadoop’},{‘hello’,’oracle’}} 相當於二維數組
這在使用Apache日誌統計PV時很有用例如日誌格式如下：
123.23.4.5 - - xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
23.12.4.5 - - xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
我們只需要取出按空格分隔的第一個列即可 這是使用flatMap就不合適了我們可以用map
Salca > var file=sc.textFile(“hdfs://h201:9000/access.log”)
Salca> var h1=file.map(_.split(“ ”,2)) #按空格分隔最多兩列
Salca> var h2=h1.map(x=>(x(0),1)) #輸入的數組去第0列，即可取出IP
Salca> var h3=h2.reduceByKey(_+_) #統計pv
下面就是排序和保存在這裏就不在重復了。
Spark shell 詞頻統計和統計PV心得