日志统计分析

    /**
      * Spark SQL 用户自定义函数
      */
    import org.apache.spark.sql.functions._
    def formatComment() = udf((comment:String) =>{
            
    })

    /**
      * 在已有DF上添加或者修改字段
      */
    outputDF = outputDF.withColumn("dis_diag_comment", formatComment()(outputDF("dis_diag_comment")))

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

/**
  * Description: 日志统计分析
  * .collect().foreach(println)
  * .take(10).foreach(println)
  *
  *
  * 分隔符 ，
  *map()是对每一行数据进行操作
  *reduceByKey()是将相同的Key分到一块去，进行+操作
  *
  * @Author: 留歌36
  * @Date: 2019/8/2 14:53
  */
object LogApp {
  def main(args: Array[String]): Unit = {
    // Spark编程模板第一步：创建SparkContext
    val conf = new SparkConf().setMaster("local[2]").setAppName("LogApp")
    val sc = new SparkContext(conf)

    // Spark编程模板第二步：读取日志文件，并进行相应的业务逻辑处理
    val path = "D:\IDEA_Work\spark-august\data\hello.txt"
    val lines = sc.textFile(path)

    // TODO ...  需求一：求各域名的流量
    /**
      *  2 字段 域名
      *  4 字段 流量
      * 1) 先获取每条日志记录下的 域名11字段 流量 20字段
      *       log  ==> (11,20)
      * 2) 按照域名进行分组求和  reduceByKey()
      *
      * 3)拓展，成功了多少条，错误了多少条？
      *
      * .saveAsTextFile()
      */
//    lines.map( x =>{
//      val temp = x.split(",")
//      var traffic = 0L  // 考虑到流量这个值可能脏数据，无法toLong
//      try{
//        traffic = temp(3).trim.toLong   // 考虑到空格的情况
//      }catch {
//        case e:Exception => traffic = 0L
//      }
//      (temp(1),traffic)
//    }).reduceByKey(_+_).take(10).foreach(println)


    // TODO ...  需求二：求各省份访问次数的TopN（次数排序）
    /**
      *  5字段IP
      * 1) 日志中获取IP，解析为省份
      * 2) (省份, 1)
      * 3) reduceByKey(_+_)
      * 4) 排序
      *
      *
      */
    lines.map(x=>{
      val temp = x.split(",")
      (IpUtils.getProvince(temp(4)), 1)
    }).reduceByKey(_+_).sortBy(_._2, false).take(10).foreach(println)

    // TODO ...  需求三：求各域名下 各个文件资源(/a/b/c/123.mp4)的访问次数TopN
    /**
      * 6 字段 文件资源
      * http://domain/a/b/c/123.mp4?x=y&r=z
      * 定义：第一个/之后到第一个?之前的内容
      *
      *
      */

//    lines.map(x=>{
//      val temp = x.split(",")
//      ((temp(1), getResource(temp(5))), 1)
//    })
//      .reduceByKey(_+_).groupBy(_._1._1)
//      .mapValues(_.toList.sortBy(_._2).reverse.take(10))
//      .collect().foreach(println)


    // TODO... 求访问次数最多的资源文件

//    lines.map(x=>{
//      val temp = x.split(",")
//      (getResource(temp(5)), 1)
//    }).reduceByKey(_+_).sortBy(_._2, false).collect().foreach(println)



    // Spark编程模板第三步:关闭SparkContext
  }

  def getResource(url :String)= {
    val pathTemp = url.replaceFirst("//","")
    var pathIndex = pathTemp.indexOf("/")
    var path =""
    if (pathIndex != -1){
      path = pathTemp.substring(pathIndex)
      pathIndex = path.indexOf("?")
      if (pathIndex != -1){
        path = path.substring(0, pathIndex)
      }
    }
    path
  }

}