Spark【案例】——PV、UV、TopK

Spark【案例】——PV、UV、TopK。

/**
  * 计算pv  log日志文件中每一行记录，为一次点击记录，也就是一次pv操作
  */
object LocalPV {

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

    val rdd1 = sc.textFile("E:/here/ceshi/access.log")
    val rdd2 = rdd1.map(x => ("pv", 1))
    val rdd3 = rdd2.reduceByKey(_+_)
    val rdd4 = rdd3.saveAsTextFile("E:/here/ceshi/result/access.log")

    sc.stop()
  }

}

/**
  * 计算UV  log文件的每一行为每一条访问记录,每条记录中第一个元素为访问的ip
  */
class LocalUV {

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

    val rdd1 = sc.textFile("E:/here/ceshi/access.log")

    // 分割每行数据取出IP，去重
    val rdd2 = rdd1.map(_.split(" ")).map(x => x(0)).distinct().map(x => ("uv", 1))
    rdd2.foreach(println)
  }

}

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

/**
  * 求topk  log日志中每一行会有一个来源网址信息，统计所有日志中访问量排序并列出其中的k个
  */
class TopK {

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

    val rdd1 = sc.textFile("E:/here/ceshi/access.log")

    val rdd2 = rdd1.map(_.split(" ")).map(x => x(10)).map(f = (_, 1))
    val result = rdd2.reduceByKey(_ + _).sortBy(_._2, false).take(3)
    println(result.toList)
  }

}