024 關於spark中日志分析案例

1.四個需求

　　需求一：求contentsize的平均值、最小值、最大值

　　需求二：請各個不同返回值的出現的數據 ===> wordCount程序

　　需求三：獲取訪問次數超過N次的IP地址

　　需求四：獲取訪問次數最多的前K個endpoint的值 ==> TopN

 

2.主程序LogAnalyzer.scala

package com.ibeifeng.bigdata.spark.core.log

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

/**
  * Apache日志分析
  * Created by ibf on 01/15.
  */
object LogAnalyzer {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("log-analyzer")
      .setMaster("local[*]")
      .set("spark.eventLog.enabled", "true")
      .set("spark.eventLog.dir", "hdfs://hadoop-senior01:8020/spark-history")
    val sc = SparkContext.getOrCreate(conf)

    // ================日志分析具體代碼==================
    // HDFS上日志存儲路徑
    val path = "/beifeng/spark/access/access.log"

    // 創建rdd
    val rdd = sc.textFile(path)

    // rdd轉換，返回進行后續操作
    val apacheAccessLog: RDD[ApacheAccessLog] = rdd
      // 過濾數據
      .filter(line => ApacheAccessLog.isValidateLogLine(line))
      .map(line => {
        // 對line數據進行轉換操作
        ApacheAccessLog.parseLogLine(line)
      })

    // 對多次時候用的rdd進行cache
    apacheAccessLog.cache()

    // 需求一：求contentsize的平均值、最小值、最大值
    /*
    * The average, min, and max content size of responses returned from the server.
    * */
    val contentSizeRDD: RDD[Long] = apacheAccessLog
      // 提取計算需要的字段數據
      .map(log => (log.contentSize))

    // 對重復使用的RDD進行cache
    contentSizeRDD.cache()

    // 開始計算平均值、最小值、最大值
    val totalContentSize = contentSizeRDD.sum()
    val totalCount = contentSizeRDD.count()
    val avgSize = 1.0 * totalContentSize / totalCount
    val minSize = contentSizeRDD.min()
    val maxSize = contentSizeRDD.max()

    // 當RDD不使用的時候，進行unpersist
    contentSizeRDD.unpersist()

    // 結果輸出
    println(s"ContentSize Avg：${avgSize}, Min: ${minSize}, Max: ${maxSize}")

    // 需求二：請各個不同返回值的出現的數據 ===> wordCount程序
    /*
    * A count of response code's returned.
    * */
    val responseCodeResultRDD = apacheAccessLog
      // 提取需要的字段數據, 轉換為key/value鍵值對，方便進行reduceByKey操作
      // 當連續出現map或者flatMap的時候，將多個map/flatMap進行合並
      .map(log => (log.responseCode, 1))
      // 使用reduceByKey函數，按照key進行分組后，計算每個key出現的次數
      .reduceByKey(_ + _)

    // 結果輸出
    println(s"""ResponseCode :${responseCodeResultRDD.collect().mkString(",")}""")

    // 需求三：獲取訪問次數超過N次的IP地址
    // 需求三額外：對IP地址進行限制，部分黑名單IP地址不統計
    /*
    * All IPAddresses that have accessed this server more than N times.
    * 1. 計算IP地址出現的次數 ===> WordCount程序
    * 2. 數據過濾
    * */
    val blackIP = Array("200-55-104-193.dsl.prima.net.ar", "10.0.0.153", "208-38-57-205.ip.cal.radiant.net")
    // 由於集合比較大，將集合的內容廣播出去
    val broadCastIP = sc.broadcast(blackIP)
    val N = 10
    val ipAddressRDD = apacheAccessLog
      // 過濾IP地址在黑名單中的數據
      .filter(log => !broadCastIP.value.contains(log.ipAddress))
      // 獲取計算需要的IP地址數據，並將返回值轉換為Key/Value鍵值對類型
      .map(log => (log.ipAddress, 1L))
      // 使用reduceByKey函數進行聚合操作
      .reduceByKey(_ + _)
      // 過濾數據，要求IP地址必須出現N次以上
      .filter(tuple => tuple._2 > N)
    // 獲取滿足條件IP地址, 為了展示方便，將下面這行代碼注釋
    //      .map(tuple => tuple._1)

    // 結果輸出
    println(s"""IP Address :${ipAddressRDD.collect().mkString(",")}""")

    // 需求四：獲取訪問次數最多的前K個endpoint的值 ==> TopN
    /*
    * The top endpoints requested by count.
    * 1. 先計算出每個endpoint的出現次數
    * 2. 再進行topK的一個獲取操作，獲取出現次數最多的前K個值
    * */
    val K = 10
    val topKValues = apacheAccessLog
      // 獲取計算需要的字段信息，並返回key/value鍵值對
      .map(log => (log.endpoint, 1))
      // 獲取每個endpoint對應的出現次數
      .reduceByKey(_ + _)
      // 獲取前10個元素, 而且使用我們自定義的排序類
      .top(K)(LogSortingUtil.TupleOrdering)
    // 如果只需要endpoint的值，不需要出現的次數，那么可以通過map函數進行轉換
    //      .map(_._1)

    // 結果輸出
    println(s"""TopK values:${topKValues.mkString(",")}""")


    // 對不在使用的rdd，去除cache
    apacheAccessLog.unpersist()

    // ================日志分析具體代碼==================

    sc.stop()
  }
}

 

3.需要的輔助類一（返回匹配的日志）

package com.ibeifeng.bigdata.spark.core.log

import scala.util.matching.Regex

/**
  * 64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846
  * Created by ibf on 01/15.
  */
case class ApacheAccessLog(
                            ipAddress: String, // IP地址
                            clientId: String, // 客戶端唯一標識符
                            userId: String, // 用戶唯一標識符
                            serverTime: String, // 服務器時間
                            method: String, // 請求類型/方式
                            endpoint: String, // 請求的資源
                            protocol: String, // 請求的協議名稱
                            responseCode: Int, // 請求返回值：比如：200、401
                            contentSize: Long // 返回的結果數據大小
                          )

/**
  * 64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846
  * on 01/15.
  * 提供一些操作Apache Log的工具類供SparkCore使用
  */
object ApacheAccessLog {
  // Apache日志的正則
  val PARTTERN: Regex =
  """^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+) (\S+) (\S+)" (\d{3}) (\d+)""".r

  /**
    * 驗證一下輸入的數據是否符合給定的日志正則，如果符合返回true；否則返回false
    *
    * @param line
    * @return
    */
  def isValidateLogLine(line: String): Boolean = {
    val options = PARTTERN.findFirstMatchIn(line)

    if (options.isEmpty) {
      false
    } else {
      true
    }
  }

  /**
    * 解析輸入的日志數據
    *
    * @param line
    * @return
    */
  def parseLogLine(line: String): ApacheAccessLog = {
    if (!isValidateLogLine(line)) {
      throw new IllegalArgumentException("參數格式異常")
    }

    // 從line中獲取匹配的數據
    val options = PARTTERN.findFirstMatchIn(line)

    // 獲取matcher
    val matcher = options.get

    // 構建返回值
    ApacheAccessLog(
      matcher.group(1), // 獲取匹配字符串中第一個小括號中的值
      matcher.group(2),
      matcher.group(3),
      matcher.group(4),
      matcher.group(5),
      matcher.group(6),
      matcher.group(7),
      matcher.group(8).toInt,
      matcher.group(9).toLong
    )
  }
}

 

4.需要的輔助類二（自定義的一個二元組的比較器，方便進行TopN）

package com.ibeifeng.bigdata.spark.core.log

/**
  * Created by ibf on 01/15.
  */
object LogSortingUtil {

  /**
    * 自定義的一個二元組的比較器
    */
  object TupleOrdering extends scala.math.Ordering[(String, Int)] {
    override def compare(x: (String, Int), y: (String, Int)): Int = {
      // 按照出現的次數進行比較，也就是按照二元組的第二個元素進行比較
      x._2.compare(y._2)
    }
  }

}