spark|六十五、Spark-综合案例(搜狗搜索日志分析) 运维|spark|big|data|运维|hadoop

搜狗实验室：搜索引擎查询日志库设计为包括约1个月(2008年6月)Sogou搜索引擎部分网页查询需求及用户点击情况的网页查询日志数据集合。为进行中文搜索引擎用户行为分析的研究者提供基准研究语料

目录
原数据展示
业务需求
业务逻辑
分词工具
Maven依赖
代码实现
效果展现

搜狗搜索日志官网：http://www.sogou.com/labs/resource/q.php

文章图片
?
迷你版日志下载链接：http://download.labs.sogou.com/dl/sogoulabdown/SogouQ/SogouQ.mini.zip

文章图片
?
注：由于进行测试使用，迷你版数据就可以满足需求

原数据展示

文章图片
?
注：原数据存在10000条，字段分别为：访问时间 \t 用户ID \t [查询词] \t 该URL在返回结果中的排名 \t 用户点击的顺序号 \t用户点击的URL

业务需求

需求说明：对SougouSearchLog进行分词并统计如下指标:

热门搜索词

用户热门搜索词(带上用户id)

各个时间段搜索热度

文章图片
?

业务逻辑

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

【spark|六十五、Spark-综合案例(搜狗搜索日志分析)】

分词工具

HanLP官网：http://www.sogou.com/labs/resource/q.php

文章图片
?

文章图片
?
HanLP主要功能：基于HanLP最新技术，使用亿级通用语料库训练，直接API调用，简单高效！

Maven依赖

com.hankcs hanlp portable-1.7.7

HanLP入门案例

package org.example.sparkimport java.utilimport com.hankcs.hanlp.HanLP import com.hankcs.hanlp.seg.common.Term/** * Author tuomasi * Desc HanLP入门案例 */ 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] import scala.collection.JavaConverters._ println(terms.asScala.map(_.word)) //转为scala的list:ArrayBuffer([, HanLP, 入门, 案例, ])val cleanWords1: String = words.replaceAll("\\[|\\]", "") //将"["或"]"替换为空"" //"HanLP入门案例" println(cleanWords1) //HanLP入门案例 println(HanLP.segment(cleanWords1).asScala.map(_.word)) //ArrayBuffer(HanLP, 入门, 案例)val log = """00:00:00 2982199073774412[360安全卫士]8 3 download.it.com.cn/softweb/software/firewall/antivirus/20067/17938.html""" val cleanWords2 = log.split("\\s+")(2) //[360安全卫士] .replaceAll("\\[|\\]", "") //360安全卫士 println(HanLP.segment(cleanWords2).asScala.map(_.word)) //ArrayBuffer(360, 安全卫士) } }

控制台打印效果

文章图片
?

代码实现

package org.example.sparkimport com.hankcs.hanlp.HanLP import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} import shapeless.record import spire.std.tuplesimport scala.collection.immutable.StringOps import scala.collection.mutable/** * Author tuomasi * Desc 需求:对SougouSearchLog进行分词并统计如下指标: * 1.热门搜索词 * 2.用户热门搜索词(带上用户id) * 3.各个时间段搜索热度 */ object SougouSearchLogAnalysis { 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/input/SogouQ.sample")//TODO 2.处理数据 //封装数据 val SogouRecordRDD: RDD[SogouRecord] = lines.map(line => { //map是一个进去一个出去 val arr: Array[String] = line.split("\\s+") SogouRecord( arr(0), arr(1), arr(2), arr(3).toInt, arr(4).toInt, arr(5) ) })//切割数据 /* val wordsRDD0: RDD[mutable.Buffer[String]] = SogouRecordRDD.map(record => { val wordsStr: String = record.queryWords.replaceAll("\\[|\\]", "") //360安全卫士 import scala.collection.JavaConverters._ //将Java集合转为scala集合 HanLP.segment(wordsStr).asScala.map(_.word) //ArrayBuffer(360, 安全卫士) })*/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(t => !t._2.equals(".") && !t._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 hourAndMitunesStr: String = timeStr.substring(0, 5) (hourAndMitunesStr, 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 ) }

效果展现