Spark项目实战—各省份广告TOP3分析

1. 数据准备
   agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
   百度链接：百度网盘数据链接
   https://pan.baidu.com/s/1cnTINRgCBencvMTw4xXXH
   提取码：nenx

2. 需求描述
   统计出每一个省份每个广告被点击数量排行的 Top3

3. 需求分析
   

4. 代码实现

package req

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

object Spark01_Operator_RDDreq {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("operatorRDD_req").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 案例实操

    //1、读取原始数据：时间戳，省份，城市，用户，广告
    val dataRDD: RDD[String] = sc.textFile("data/agent.log")

    //2、将原始数据进行结构转换，方便统计
    // 时间戳，省份，城市，用户，广告 => ((省份，广告)，1)
    val mapRDD: RDD[((String, String), Int)] = dataRDD.map(
      line => {
        val datas = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    )

    //3、将转换结构的数据，进行分组聚合
    // ((省份,广告),1) => ((省份,广告),sum)
    val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(_ + _)
    //4、将聚合的结果进行结构的转换
    //((省份,广告),sum) => ((省份,(广告,sum)) ,使用模式匹配的方式

    val newmapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case ((prv, ads), sum) => {
        (prv, (ads, sum))
      }
    }

    //5、将转换结构后的数据根据省份进行分组
    //(省份,[(广告A,sumA)],[(广告B,sumB)])
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newmapRDD.groupByKey()

    //6、将分组后的数据组内排序(降序)，取前3名
    // Iterable可迭代的集合不能直接排序，需将其转换为List
    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3) //Iterable[(String, Int)]根据Int的数值来排序，是_._2
      }
    )

    //7.采集数据打印在控制台
    resultRDD.collect().foreach(println)

    

  }
}