Spark流计算课程设计----帮助新手入门实践

红太狼 2023-05-28 15:50 11阅读 0赞

### **一、概述** ###

**1.1 问题背景**

某网站系统实时产生日志信息，记录用户对系统的访问信息，例如：IP地址，用户名称，访问时间，请求和响应信息，其中IP地址信息是表示全国各地用户的访问情况，对IP地址的详细分析，可以了解各个地区对该网站系统访问的活跃度，用以判断该网站公司对区域活动的推广情况和投入成本。本项目为网站运营方向常用技术案例。

**1.2 本组完成的任务**

1.  对日志进行清洗。
2.  统计24小时内的访问量Top10。
3.  统计该时间段内的PV。
4.  统计该时间段内的UV。
5.  统计该时间段跳出用户信息
6.  将获取到的结果存为本地文件result，并进行了可视化。

**1.3 数据下载地址**

[https://download.csdn.net/download/zhiyeegao/12251446][https_download.csdn.net_download_zhiyeegao_12251446]

### **二、数据格式分析** ###

所给的数据文件名为access.log。其中每列的内容如下：

第一列：客户端用户的IP地址。

第二列：客户端用户名称，默认为- -。

第三列：客户端访问的时间与时区。例如：\[04/Jan/2012:19:57:09 +0800\]代表中国东8区2012年1月4号19点57分9秒。

第四列：记录请求的url和http协议。例如：GET /thread-1459739-1-1.html HTTP/1.1

第五列：记录请求状态，成功是200。

第六列：记录发给客户端的字节数。

第七列：记录从哪个页面链接访问过来的。

第八列：记录客户端浏览器相关信息。

**三、数据处理方案**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70][]

图-1 数据处理方案图

### **四、源码（scala）** ###

import java.text._
    import java.util._
    import java.lang._
    import java.io._
    import org.apache.spark._
    import scala.util.control._
    object Clean {
      /**
      * 自定义数据预处理函数
      * @param line:String (line是access.log文件的每一行)
      * @return (IP+" "+d1)
      */
      def find_IP_Date(line:String):String={
        var IP:String = "";//字符串IP存储IP地址
        var d1:String = "";//字符串d1存储时间和时区
        try{
    	//分割字符串得到IP字符串
        IP = line.split(" - ")(0).trim()
    	//日期字符串的起点
        val left = line.indexOf("[")
    	//日期字符串的终点
        val right = line.indexOf("]")
    	//截取字符串得到日期字符串
        val date = line.substring(left + 1, right).trim()
        d1 = date
        }catch{
          case e:Exception=>println(e)//输出异常
        }
    	//返回IP和日期
        return (IP+" "+d1)
      }
      /**
      * 自定义跳出用户判断函数
      * @param line:Array[Int] (line中是排好序的访问时间)
      * @return flag:Boolean
      */
      def choose(line:Array[Int]):Boolean={
    	//flag标记选择结果
        var flag:Boolean = false
        for(i <- 1 until line.length){
    	//如果时差大于1小时则认为用户跳出
          if(line(i)-line(i-1)>1)
            flag = true
        }
        return flag
      }
    
      def main(args:Array[String]){
    	//创建文件输出流，将结果写入result文件
        val writer = new PrintWriter(new File("/root/result"))
    	//创建SparkConf
        val conf = new SparkConf().setMaster("local[2]").setAppName("IP_Date_Clean")
    	//创建SparkContext
        val sc = new SparkContext(conf);
    	//创建lines RDD
        val lines = sc.textFile("/root/access.log")
    	//数据预处理
        val words = lines.map(line=>find_IP_Date(line)).distinct()
        //words.take(10).foreach(println) //查看前10个，进行输出测试
          
        //统计PV
        println("total pv = " + words.count())//pv总数量等于日志的行数
        writer.println("total pv = " + words.count())//写入文件
    	//创建hour_pv RDD存放每小时的pv数量
        val hour_pv = words.map(line=>(line.split(":")(1).toInt,1)).reduceByKey((a,b)=>a+b)
        println("one hour pv:")
        writer.println("one hour pv:")//写入文件
        hour_pv.foreach(println) //输出每小时的pv数量
        hour_pv.collect.map{line=>writer.println(line)}//写入文件
        
        //统计UV
    	//创建ips RDD存放所有的IP地址
        val ips = words.map(line => line.split(" ")(0))
    	//创建dips RDD存放去重后的IP地址
        val dips = ips.distinct()
        println("uv = " + dips.count())
        writer.println("uv = " + dips.count().toString())//写入文件
    
        //统计top10的IP (就是简单的IP计数和排序，然后取前10个)
        val pairs = ips.map { ip  => (ip,1) }
        val ipCounts = pairs.reduceByKey(_+_)
        val countips = ipCounts.map(ipCount  => (ipCount._2,ipCount._1))
        val sortedCountips = countips.sortByKey(false)
        val sortedipCounts = sortedCountips.map(sortedCountip =>(sortedCountip._2,sortedCountip._1))
        println("top 10 ip:")
        writer.println("top 10 ip:")//写入文件
        sortedipCounts.take(10).foreach(println)
        sortedipCounts.take(10).map{line=>writer.println(line.toString())}//写入文件
        
        //统计跳出数
        //time_interval等于1代表以一个小时作为用户跳出判断标准
        val time_interval = 1
    	//创建time RDD存放IP和其对应小时
        val time = words.map(line => (line.split(" ")(0),line.split(":")(time_interval).toInt))
    	//创建ip_time RDD对time进行分组，并存放IP和其对应时间数组
        val ip_time = time.groupByKey().map(line => (line._1,line._2.toArray.sorted))
        //ip_time.take(2).map(line => line._2.foreach(println)) //查看前2个，进行输出测试
    	//创建jump_ip RDD存放跳出用户
        val jump_ip = ip_time.filter(line => choose(line._2))
        println("jump = " + jump_ip.count())
        writer.println("jump = " + jump_ip.count())//写入文件
        println("jump probability = " + jump_ip.count()*100.0/dips.count()+"%")
        writer.println("jump probability = " + jump_ip.count()*100.0/dips.count()+"%")//写入文件
        writer.close()//关闭文件输出流
      }  
    }

### **五、总结** ###

**通过分析课设要求和日志文件，我主要使用了IP和time两个字段，围绕这两个字段的数据进行了一系列的RDD操作，提取了PV、UV、访问量TOP10和跳出数、跳出率等数据。**

**其中我遇到了两个问题：**

**(1)****问题：在提取IP和time字段时，程序运行中断了。**

**原因：文件中有一部分噪音数据，在提取字段时遇到异常，程序中止。**

**解决方法：加上了try-catch，跳过异常就可以正常运行了。**

**(2)****问题：在对(IP,Iterable)中的Iterable进行排序时报错。**

**原因：我对Iterable的排序方法不熟悉。**

**解决方法：将Iterable转换为Array。**

### **六、结果展示** ###

total pv = 344129  
one hour pv:  
(0,41)  
(20,79107)  
(21,90885)  
(22,89733)  
(19,3780)  
(23,80583)  
uv = 8549  
top 10 ip:  
(114.112.141.6,9364)  
(113.31.47.158,7817)  
(199.255.44.5,5744)  
(111.13.8.250,4353)  
(61.177.46.238,4124)  
(203.80.118.166,2775)  
(61.135.248.199,2626)  
(124.16.186.16,2232)  
(218.25.117.235,1936)  
(203.208.60.175,1913)  
jump\_hour = 439  
jump\_hour probability = 5.135103520879635%  
jump\_minute = 3352  
jump\_minute probability = 39.20926424143175%  
jump\_second = 6282  
jump\_second probability = 73.48227862907943%

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70 2][]

[https_download.csdn.net_download_zhiyeegao_12251446]: https://download.csdn.net/download/zhiyeegao/12251446
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70]: /images/20230528/b53e16d965ae4d5eaf65700ae0b1dbf1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70 1]: /images/20230528/5a6ab4190b7e477385a3c968747da3c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXllZWdhbw_size_16_color_FFFFFF_t_70 2]: /images/20230528/c6107b7406ca44e88d98de713df78d8d.png