关于大数据:大数据开发SparkRDD实操案例http日志分析

10次阅读

共计 7546 个字符，预计需要花费 19 分钟才能阅读完成。

配置文件，或者配置表，个别是放在在线 db，比方 mysql 等关系型数据库，或者后盾 rd 间接丢给你一份文件，数据量比起整个离线数据仓库的大表来说算很小，所以这种状况下，个别的做法是将小表，或者小文件播送进来，那么上面一个例子来看，播送表的应用解决 ip 地址映射问题

数据地址：链接：https://pan.baidu.com/s/1FmFxSrPIynO3udernLU0yQ 提取码：hell

http.log：

用户拜访网站所产生的日志。日志格局为：工夫戳、IP 地址、拜访网址、拜访数据、浏览器信息等，样例如下：

ip.dat：ip 段数据，记录着一些 ip 段范畴对应的地位，总量大略在 11 万条，数据量也算很小的，样例如下

文件地位：data/http.log、data/ip.dat

链接：https://pan.baidu.com/s/1FmFxSrPIynO3udernLU0yQ 提取码：hell

要求：将 http.log 文件中的 ip 转换为地址。如将 122.228.96.111 转为温州，并统计各城市的总访问量

有三个关键点，http.log 的要害信息是 ip 地址，所以依据数据的精简准则，只读取 ip 即可，另外 ip 映射比对的时候，ip 地址映射文件是排序的，所以为了进步查找效率，采纳将 ip 地址转为 long 类型，而后再用二分法来查找，找到地址后映射为地址。

package com.hoult.work

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * 数据源：1.ip 地址的拜访日志 2.ip 地址映射表
 * 须要把映射表播送，地址转换为 long 类型进行比拟
 */
object FindIp {def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getCanonicalName)
      .getOrCreate()
    val sc = spark.sparkContext

    import spark.implicits._
    val ipLogsRDD = sc.textFile("data/http.log")
      .map(_.split("\\|")(1))


    val ipInfoRDD = sc.textFile("data/ip.dat").map {
      case line: String => {val strSplit: Array[String] = line.split("\\|")
        Ip(strSplit(0), strSplit(1), strSplit(7))
      }
    }


    val brIPInfo = sc.broadcast(ipInfoRDD.map(x => (ip2Long(x.startIp), ip2Long(x.endIp), x.address))collect())

    // 关联后的后果 rdd
    ipLogsRDD
      .map(x => {val index  = binarySearch(brIPInfo.value, ip2Long(x))
        if (index != -1)
          brIPInfo.value(index)._3
        else
          "NULL"
      }).map(x => (x, 1))
      .reduceByKey(_ + _)
      .map(x => s"城市：${x._1}, 访问量：${x._2}")
      .saveAsTextFile("data/work/output_ips")

  }

  //ip 转成 long 类型
  def ip2Long(ip: String): Long = {val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length) {ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }

  // 二分法匹配 ip 规定
  def binarySearch(lines: Array[(Long, Long, String)], ip: Long): Int = {
    var low = 0
    var high = lines.length - 1
    while (low <= high) {val middle = (low + high) / 2
      if ((ip >= lines(middle)._1) && (ip <= lines(middle)._2))
        return middle
      if (ip < lines(middle)._1)
        high = middle - 1
      else {low = middle + 1}
    }
    -1
  }

}

case class Ip(startIp: String, endIp: String, address: String)

后果截图如下：

日志格局 ：IP 命中率(Hit/Miss) 响应工夫申请工夫申请办法申请 URL 申请协定状态码响应大小 referer 用户代理

日志文件地位：data/cdn.txt

数据 case:

工作：

2.1、计算独立 IP 数

2.2、统计每个视频独立 IP 数（视频的标记：在日志文件的某些能够找到 *.mp4，代表一个视频文件）

2.3、统计一天中每个小时的流量

剖析：刚开始去找格林工夫的 jod-time 解析，找了一圈不晓得该怎么写，前面发现只须要小时即可，应用正则来提取, 留神在求 video 的拜访 ip 时候，能够用aggregateByKey 来进步性能

package com.hoult.work

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * 读取日志表到 rdd
 * 拿到须要的字段：ip, 拜访工夫：小时即可, 视频名 video_name (url 中的 xx.mp4),
 * 剖析：* 1. 计算独立 IP 数
 * 2. 统计每个视频独立 IP 数（视频的标记：在日志文件的某些能够找到 *.mp4，代表一个视频文件）* 3. 统计一天中每个小时的流量
 */
object LogAnaylse {def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getCanonicalName)
      .getOrCreate()
    val sc = spark.sparkContext


    val cdnRDD = sc.textFile("data/cdn.txt")

    // 计算独立 ips
//    aloneIPs(cdnRDD.repartition(1))

    // 每个视频独立 ip 数
//    videoIPs(cdnRDD.repartition(1))

    // 每小时流量
    hourPoor(cdnRDD.repartition(1))
  }



  /**
   * 独立 ip 数
   */
  def aloneIPs(cdnRDD: RDD[String]) = {
    // 匹配 ip 地址
    val IPPattern = "((?:(?:25[0-5]|2[0-4]\\d|((1\\d{2})|([1-9]?\\d)))\\.){3}(?:25[0-5]|2[0-4]\\d|((1\\d{2})|([1-9]?\\d))))".r

    val ipnums = cdnRDD
      .flatMap(x => (IPPattern findFirstIn x))
      .map(y => (y,1))
      .reduceByKey(_+_)
      .sortBy(_._2,false)

    ipnums.saveAsTextFile("data/cdn/aloneIPs")
  }

  /**
   * 视频独立 ip 数
   */
  def videoIPs(cdnRDD: RDD[String]) = {
    // 匹配 http 响应码和申请数据大小
    val httpSizePattern = ".*\\s(200|206|304)\\s([0-9]+)\\s.*".r


    //[15/Feb/2017:11:17:13 +0800]  匹配 2017:11 按每小时播放量统计
    val timePattern = ".*(2017):([0-9]{2}):[0-9]{2}:[0-9]{2}.*".r

    import scala.util.matching.Regex

    // Entering paste mode (ctrl-D to finish)

    def isMatch(pattern: Regex, str: String) = {
      str match {case pattern(_*) => true
        case _ => false
      }
    }

    def getTimeAndSize(line: String) = {var res = ("", 0L)
      try {val httpSizePattern(code, size) = line
        val timePattern(year, hour) = line
        res = (hour, size.toLong)
      } catch {case ex: Exception => ex.printStackTrace()
      }
      res
    }

    val IPPattern = "((?:(?:25[0-5]|2[0-4]\\d|((1\\d{2})|([1-9]?\\d)))\\.){3}(?:25[0-5]|2[0-4]\\d|((1\\d{2})|([1-9]?\\d))))".r

    val videoPattern = "([0-9]+).mp4".r

    val res = cdnRDD
      .filter(x => x.matches(".*([0-9]+)\\.mp4.*"))
      .map(x => (videoPattern findFirstIn x toString,IPPattern findFirstIn x toString))
      .aggregateByKey(List[String]())((lst, str) => (lst :+ str),
        (lst1, lst2) => (lst1 ++ lst2)
      )
      .mapValues(_.distinct)
      .sortBy(_._2.size,false)

      res.saveAsTextFile("data/cdn/videoIPs")
  }

  /**
   * 一天中每个小时的流量
   *
   */
  def hourPoor(cdnRDD: RDD[String]) = {val httpSizePattern = ".*\\s(200|206|304)\\s([0-9]+)\\s.*".r
    val timePattern = ".*(2017):([0-9]{2}):[0-9]{2}:[0-9]{2}.*".r
    import scala.util.matching.Regex

    def isMatch(pattern: Regex, str: String) = {
      str match {case pattern(_*) => true
        case _ => false
      }
    }

    def getTimeAndSize(line: String) = {var res = ("", 0L)
      try {val httpSizePattern(code, size) = line
        val timePattern(year, hour) = line
        res = (hour, size.toLong)
      } catch {case ex: Exception => ex.printStackTrace()
      }
      res
    }

    cdnRDD
      .filter(x=>isMatch(httpSizePattern,x))
      .filter(x=>isMatch(timePattern,x))
      .map(x=>getTimeAndSize(x))
      .groupByKey()
      .map(x=>(x._1,x._2.sum))
      .sortByKey()
      .map(x=>x._1+"时 CDN 流量 ="+x._2/(102424*1024)+"G")
      .saveAsTextFile("data/cdn/hourPoor")
  }
}

运行后果截图：

假如点击日志文件 (click.log) 和曝光日志 imp.log, 中每行记录格局如下

// 点击日志
INFO 2019-09-01 00:29:53 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=31
INFO 2019-09-01 00:30:31 requestURI:/click?app=2&p=1&adid=18005472&industry=469&adid=31
INFO 2019-09-01 00:31:03 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=32
INFO 2019-09-01 00:31:51 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=33

// 曝光日志
INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=31
INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=31
INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=34

用 Spark-Core 实现统计每个 adid 的曝光数与点击数，思路较简略，间接上代码

代码：

package com.hoult.work

import org.apache.spark.sql.SparkSession

object AddLog {def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getCanonicalName)
      .getOrCreate()
    val sc = spark.sparkContext

    val clickRDD = sc.textFile("data/click.log")
    val impRDD = sc.textFile("data/imp.log")

    val clickRes = clickRDD.map{line => {val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }}.reduceByKey(_ + _)

    val impRes = impRDD.map { line =>
      val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }.reduceByKey(_ + _)

    // 保留到 hdfs
    clickRes.fullOuterJoin(impRes)
      .map(x => x._1 + "," + x._2._1.getOrElse(0) + "," + x._2._2.getOrElse(0))
      .repartition(1)
//      .saveAsTextFile("hdfs://linux121:9000/data/")
      .saveAsTextFile("data/add_log")

    sc.stop()}
}

剖析：共有两次 shuffle, fulljon 能够批改为 union + reduceByKey，将shuffle 缩小到一次

A 表有三个字段：ID、startdate、enddate，有 3 条数据：

1 2019-03-04 2020-02-03

2 2020-04-05 2020-08-04

3 2019-10-09 2020-06-11

写 SQL（须要 SQL 和 DSL）将以上数据变动为：

2019-03-04 2019-10-09

2019-10-09 2020-02-03

2020-02-03 2020-04-05

2020-04-05 2020-06-11

2020-06-11 2020-08-04

2020-08-04 2020-08-04

剖析：察看，能够失去，第一列实际上是 startdate 和 enddate 两列叠加的后果，而第二列是下一个，能够用 lead

窗口函数

代码如下：

package com.hoult.work

import org.apache.spark.sql.{DataFrame, SparkSession}

object DataExchange {def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("DateSort")
      .master("local[*]")
      .getOrCreate()
    spark.sparkContext.setLogLevel("warn")

    // 原数据
    val tab = List((1, "2019-03-04", "2020-02-03"),(2, "2020-04-05", "2020-08-04"),(3, "2019-10-09", "2020-06-11"))
    val df: DataFrame = spark.createDataFrame(tab).toDF("ID", "startdate", "enddate")

    val dateset: DataFrame = df.select("startdate").union(df.select("enddate"))
    dateset.createOrReplaceTempView("t")

    val result: DataFrame = spark.sql(
      """
        |select tmp.startdate, nvl(lead(tmp.startdate) over(partition by col order by tmp.startdate), startdate) enddate from
        |(select "1" col, startdate from t) tmp
        |""".stripMargin)

    result.show()}

}

运行后果：

吴邪，小三爷，混迹于后盾，大数据，人工智能畛域的小菜鸟。
更多请关注

正文完

大数据

发表至：大数据

2021-01-31

0

关于大数据:对于企业所离不开的大数据服务华为云出福利了

关于大数据:强强联合ByteHouse-携手亚马逊云科技新一代云数仓服务重磅升级

关于大数据:云音乐数据资产化建设的思考与实践

关于大数据:新起点大数据分布式可视化的-DAG-任务调度系统-Taier-正式发布14版本

关于linux:XFCE史上最好用的linux-DEdesktop-environment桌面环境

关于大数据:大数据开发SparkRDD实操案例http日志分析

1. 在生产环境下，如何解决配置文件 && 表的数据处理

2. 日志剖析案例 1

2.1 数据阐明

2.2. 实现思路和代码如下

3. 日志剖析案例 2

3.1 数据阐明

3.2 实现代码

4. 广告曝光剖析案例

5. 应用 spark-sql 实现上面的转换

站内搜索