Spark综合学习笔记（三）搜狗搜索日志分析

学习致谢：

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

一、 数据

数据网站： http: //www.sogou.com/labs/resource/q.php

二、需求

针对SougoQ用户查询日志数据中不同字段，使用SparkContext读取日志数据，封装到RDD数据集中，调用Transformation函数和Action函数进行处理不同业务统计分析

三、分词工具测试

使用比较流行好用的中文分区：HanLP,面向生产环境的自然语言处理工具包，HanLP是由一系列模型与算法组成的Java工具包，目标是普及自然语言处理在生产环境中的应用
官方网站：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)//对词进行分段      println(terms)//直接打印java的List:[[/w，HanLP/nx，入门/vn，案例/n，]/w]      println(terms.asScala.map(_.word))//转为scaLa的List:ArrayBuffer([，HanLP，入门，案例，])      val cleanwords1: String = words.replaceAll("HM[/NN]","")//将"["或"]"替换为空""//"HanLP入门案例"      println(cleanwords1)//HanLP入门案例      println(HanLP.segment(cleanwords1).asScala.map(_.word))//ArrayBuffer (HanLP,入门,案例)      val log = """e:00:00 2982199073774412 [360安全卫士] 8 3 download.it.com.cn/softweb/software/firewall/antivirus/20036/179"""      val cleanwords2 = log.split("Ils+")(2)//7[360安全卫士]      println(HanLP.segment(cleanwords2).asScala.map(_.word))//ArrayBuffer(360,安全卫士)    }  }}

运行结果

四、代码实现

import com.hankcs.hanlp.HanLPimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable/**  * Author  * DESC 需求：对SougouSearchLog进行分词并统计如下指标  * 1.热门搜索词  * 2.用户热门搜索词（带上用户id）  * 3.各个时间段搜索热度  */object SouGouSearchAnalysis {  def main(args: Array[String]): Unit = {    //TODO 0.准备环境    val conf: SparkConf=new SparkConf().setAppName("spark").setMaster("local[*]")    val sc: SparkContext=new SparkContext(conf)    sc.setLogLevel("WARN")    //TODO 1.加载数据    val lines:RDD[String]=sc.textFile("data/SogouQ.sample")    //TODO 2.处理数据    //封装数据    val  SogouRecordRDD: RDD[SogouRecord]=lines.map(line=>{//map是一个进去一个出去        var arr: Array[String]=line.split("//s+")      SogouRecord(        arr(0),        arr(1),        arr(2),        arr(3).toInt,        arr(4).toInt,        arr(5)      )  })    //切割数据    val wordsRDD:RDD[String]=SogouRecordRDD.flatMap(record=>{ //flatmap是一个进去多个出去会被压扁 //360安全卫士==》[360,安全卫士]      val wordsStr:String =record.queryWords.replaceAll("//[|//]","")//360安全卫士      import scala.collection.JavaConverters._//将Java集合转为Scala集合      HanLP.segment(wordsStr).asScala.map(_.word)//ArrayBuffer(360,安全卫士)    })    //TODO 3.统计指标    //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是一个进去多个出去会被压扁 //360安全卫士==》[360,安全卫士]      val wordsStr:String =record.queryWords.replaceAll("//[|//]","")//360安全卫士      import scala.collection.JavaConverters._//将Java集合转为Scala集合      val words: mutable.Buffer[String]=HanLP.segment(wordsStr).asScala.map(_.word)//ArrayBuffer(360,安全卫士)      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.各个时间段搜索热度    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.输出结果    result1.foreach(println)    result2.foreach(println)    result3.foreach(println)    //TODO 5.释放资源      sc.stop()  }//准备一个样例类用来封装数据}/**用户搜索点击网页记录Record  *@param queryTime 访问时间，格式为：HH:mm:ss  *@param userId     用户ID  *@param queryWords 查询词  *@param resultRank 该URL在返回结果中的排名  *@param clickRank  用户点击的顺序号  *@param clickUrl   用户点击的URL  */case class SogouRecord(                        queryTime:String,                        userId:String,                        queryWords:String,                        resultRank:Int,                        clickRank:Int,                        clickUrl:String                      )