學(xué)習(xí)致謝：

https://www.bilibili.com/video/BV1Xz4y1m7cv?p=36

一、 數(shù)據(jù)

數(shù)據(jù)網(wǎng)站： http: //www.sogou.com/labs/resource/q.php

二、需求

針對(duì)SougoQ用戶查詢?nèi)罩緮?shù)據(jù)中不同字段，使用SparkContext讀取日志數(shù)據(jù)，封裝到RDD數(shù)據(jù)集中，調(diào)用Transformation函數(shù)和Action函數(shù)進(jìn)行處理不同業(yè)務(wù)統(tǒng)計(jì)分析

三、分詞工具測試

使用比較流行好用的中文分區(qū)：HanLP,面向生產(chǎn)環(huán)境的自然語言處理工具包，HanLP是由一系列模型與算法組成的Java工具包，目標(biāo)是普及自然語言處理在生產(chǎn)環(huán)境中的應(yīng)用
官方網(wǎng)站：http://www.hanlp.com/ 添加maven依賴

<dependency><groupId>com.hankcsgroupId><artifactId>hanlpartifactId><version>portable-1.7.7version>dependency>

import com.hankcs.hanlp.HanLPimport com.hankcs.hanlp.seg.common.Termimport scala.collection.JavaConverters._object HanLPTest {  object HanLPTest {    def main(args: Array[String]):Unit = {      val words = "[HanLP入門案例]"      val terms: util.List[Term] = HanLP.segment(words)//對(duì)詞進(jìn)行分段      println(terms)//直接打印java的List:[[/w，HanLP/nx，入門/vn，案例/n，]/w]      println(terms.asScala.map(_.word))//轉(zhuǎn)為scaLa的List:ArrayBuffer([，HanLP，入門，案例，])      val cleanwords1: String = words.replaceAll("HM[/NN]","")//將"["或"]"替換為空""http://"HanLP入門案例"      println(cleanwords1)//HanLP入門案例      println(HanLP.segment(cleanwords1).asScala.map(_.word))//ArrayBuffer (HanLP,入門,案例)      val log = """e:00:00 2982199073774412 [360安全衛(wèi)士] 8 3 download.it.com.cn/softweb/software/firewall/antivirus/20036/179"""      val cleanwords2 = log.split("Ils+")(2)//7[360安全衛(wèi)士]      println(HanLP.segment(cleanwords2).asScala.map(_.word))//ArrayBuffer(360,安全衛(wèi)士)    }  }}

運(yùn)行結(jié)果

四、代碼實(shí)現(xiàn)

import com.hankcs.hanlp.HanLPimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable/**  * Author  * DESC 需求：對(duì)SougouSearchLog進(jìn)行分詞并統(tǒng)計(jì)如下指標(biāo)  * 1.熱門搜索詞  * 2.用戶熱門搜索詞（帶上用戶id）  * 3.各個(gè)時(shí)間段搜索熱度  */object SouGouSearchAnalysis {  def main(args: Array[String]): Unit = {    //TODO 0.準(zhǔn)備環(huán)境    val conf: SparkConf=new SparkConf().setAppName("spark").setMaster("local[*]")    val sc: SparkContext=new SparkContext(conf)    sc.setLogLevel("WARN")    //TODO 1.加載數(shù)據(jù)    val lines:RDD[String]=sc.textFile("data/SogouQ.sample")    //TODO 2.處理數(shù)據(jù)    //封裝數(shù)據(jù)    val  SogouRecordRDD: RDD[SogouRecord]=lines.map(line=>{//map是一個(gè)進(jìn)去一個(gè)出去        var arr: Array[String]=line.split("http://s+")      SogouRecord(        arr(0),        arr(1),        arr(2),        arr(3).toInt,        arr(4).toInt,        arr(5)      )  })    //切割數(shù)據(jù)    val wordsRDD:RDD[String]=SogouRecordRDD.flatMap(record=>{ //flatmap是一個(gè)進(jìn)去多個(gè)出去會(huì)被壓扁 //360安全衛(wèi)士==》[360,安全衛(wèi)士]      val wordsStr:String =record.queryWords.replaceAll("http://[|//]","")//360安全衛(wèi)士      import scala.collection.JavaConverters._//將Java集合轉(zhuǎn)為Scala集合      HanLP.segment(wordsStr).asScala.map(_.word)//ArrayBuffer(360,安全衛(wèi)士)    })    //TODO 3.統(tǒng)計(jì)指標(biāo)    //1.熱門搜索詞    val result1: Array[(String,Int)]=wordsRDD      .filter(word=> !word.equals(".")&& !word.equals("+"))      .map((_,1))        .reduceByKey(_+_)        .sortBy(_._2,false)        .take(10)    // 2.用戶熱門搜索詞（帶上用戶id）    val userIdAndWordRDD:RDD[(String,String)]=SogouRecordRDD.flatMap(record=>{ //flatmap是一個(gè)進(jìn)去多個(gè)出去會(huì)被壓扁 //360安全衛(wèi)士==》[360,安全衛(wèi)士]      val wordsStr:String =record.queryWords.replaceAll("http://[|//]","")//360安全衛(wèi)士      import scala.collection.JavaConverters._//將Java集合轉(zhuǎn)為Scala集合      val words: mutable.Buffer[String]=HanLP.segment(wordsStr).asScala.map(_.word)//ArrayBuffer(360,安全衛(wèi)士)      val userId: String=record.userId      words.map(word=>(userId,word))    })    val result2: Array[((String,String),Int)]=userIdAndWordRDD      .filter(word=> !word._2.equals(".")&& !word._2.equals("+"))      .map((_,1))      .reduceByKey(_+_)      .sortBy(_._2,false)      .take(10)    // 3.各個(gè)時(shí)間段搜索熱度    val result3: Array[(String,Int)]=SogouRecordRDD.map(record=>{      val timeStr:String=record.queryTime      val hourAndMinunesStr:String =timeStr.substring(0,5)      (hourAndMinunesStr,1)    }).reduceByKey(_+_)        .sortBy(_._2,false)        .take(10)    //TODO 4.輸出結(jié)果    result1.foreach(println)    result2.foreach(println)    result3.foreach(println)    //TODO 5.釋放資源      sc.stop()  }//準(zhǔn)備一個(gè)樣例類用來封裝數(shù)據(jù)}/**用戶搜索點(diǎn)擊網(wǎng)頁記錄Record  *@param queryTime 訪問時(shí)間，格式為：HH:mm:ss  *@param userId     用戶ID  *@param queryWords 查詢?cè)~  *@param resultRank 該URL在返回結(jié)果中的排名  *@param clickRank  用戶點(diǎn)擊的順序號(hào)  *@param clickUrl   用戶點(diǎn)擊的URL  */case class SogouRecord(                        queryTime:String,                        userId:String,                        queryWords:String,                        resultRank:Int,                        clickRank:Int,                        clickUrl:String                      )