使用Spark进行搜狗日志分析实例——列出搜索不同关键词超过10个的用户及其搜索的关键词

 package sogolog

 import org.apache.hadoop.io.{LongWritable, Text}

 import org.apache.hadoop.mapred.TextInputFormat

 import org.apache.spark.rdd.RDD

 import org.apache.spark.{SparkConf, SparkContext}

 class RddFile {

   def readFileToRdd(path: String): RDD[String] = {

     val conf = new SparkConf().setMaster("local").setAppName("sougoDemo")

     val sc = new SparkContext(conf);

     //使用这种方法能够避免中文乱码

     sc.hadoopFile("J:\\scala\\workspace\\first-spark-demo\\sougofile\\SogouQ.reduced",classOf[TextInputFormat], classOf[LongWritable], classOf[Text]).map{

       pair =>  new String(pair._2.getBytes, 0, pair._2.getLength, "GBK")}

   }

 }

 package sogolog

 import org.apache.spark.rdd.RDD

 /**

   * 列出搜索不同关键词超过3个的用户及其搜索的关键词

   */

 object userSearchKeyWordLT3 {

   def main(args: Array[String]): Unit = {

     //1、读入文件

     val textFile = new RddFile().readFileToRdd("J:\\scala\\workspace\\first-spark-demo\\sougofile\\SogouQ.reduced")

     //2、map操作，将每行的用户、关键词读入新的RDD中

     val userKeyWordTuple:RDD[(String,String)] = textFile.map(line=>{

       val arr = line.split("\t")

       (arr(1),arr(2))

     })

     //3、reduce操作，将相同用户的关键词进行合并

     val userKeyWordReduced = userKeyWordTuple.reduceByKey((x,y)=>{

       //去重

       if(x.contains(y)){

         x

       }else{

         x+","+y

       }

     })

     //4、使用filter进行最终过滤

     val finalResult = userKeyWordReduced.filter(x=>{

       //过滤小于10个关键词的用户

       x._2.split(",").length>=10

     })

     //5、打印出结果

     finalResult.collect().foreach(println)

   }

 }

运行结果：

使用Spark进行搜狗日志分析实例——列出搜索不同关键词超过10个的用户及其搜索的关键词的相关教程结束。