利用Spark通过nginx日志离线统计网站每日pv

#利用Spark通过nginx日志离线统计网站每日pv| 来源: 网络整理| 查看: 265

摘要

本文将介绍通过Apache Spark实现离线统计网站每日pv的思路及代码。

前言

在此之前，利用mapreduce实现了一版通过nginx日志离线分析网站每日pv，感兴趣的可以去看一下。本文实现思路与之前mapreduce的思路一致。可以很好的比较mapreduce和Spark的写法。在个人看来，Spark写起来更加优美简洁，有一种四两拨千斤的感觉。

想了解实现思路的，可以看一下利用Mapreduce实现的文章，详细思路已经阐述。

点击查看->利用HadoopMareduce实现pv统计分析

本文与Hadoop Mapreduce采用的数据集为同一个，为标准的nginx日志文件。在上文中已经提供了下载附件。感兴趣的可以去下载。

代码实现package com.itunic.rdd import java.text.SimpleDateFormat import java.util.{Date, Locale} import org.apache.spark.{SparkConf, SparkContext} import scala.collection.mutable /** * Created by c on 2017/1/11. * 通过nginx日志统计每日pv，并按照日期和pv排序 * by me: * 我本沉默是关注互联网以及分享IT相关工作经验的博客， * 主要涵盖了操作系统运维、计算机编程、项目开发以及系统架构等经验。 * 博客宗旨：把最实用的经验，分享给最需要的你， * 希望每一位来访的朋友都能有所收获！ * */ object NginxLogPV { /** * 设置需要统计的页面 */ val pages = new mutable.HashSet[String]() pages += ".php" /** * 封装KPI实体类 * * @param line * @return KPI */ def parser(line: String): KPI = { // val fields = line.split(" ") val remote_addr = fields(0) val time_local = fields(3).substring(1) val request = fields(6) val status = fields(8) var valid = true if (fields.length = 400) false else true } val url = if (request.indexOf("?") != -1) request.substring(0, request.indexOf("?")) else request KPI(remote_addr, time_local, url, status, valid) } /** * 过滤无效数据 * * @param line * @return */ def filterPVs(line: String): KPI = { val kpi: KPI = parser(line) /** * 过滤需要统计的URL */ kpi.valid = false for (page { /** * 封装并过滤数据 */ filterPVs(x) }).filter(x => { /** * 过滤有效数据 */ x.valid }).map(x => { /** * 封装 key-value数据 */ ((getTime_local_day(x.time_local), x.request), 1) }).reduceByKey(_ + _) //聚合 /** * 二次排序 */ val rdd6 = rdd.sortBy(x => PVSort(x._1._1, x._2)) /** * 格式化数据并输出到磁盘 */ rdd6.map(x => { x._1._1 + "\t" + x._1._2 + "\t" + x._2 }).saveAsTextFile("F:\\test\\input\\wc231") // println(rdd5.collect().toBuffer) sc.stop() } } /** * 自定义排序，日期升序，点击量降序 * * @param date * @param count */ case class PVSort(date: String, count: Int) extends Ordered[PVSort] with Serializable { override def compare(that: PVSort): Int = { val i = this.date.compareTo(that.date) if (i == 0) { return -this.count.compareTo(that.count) } else { return i } } } /** * kpi样例类 * * @param remote_addr * @param time_local * @param request * @param status * @param valid */ case class KPI( remote_addr: String, //来访ip time_local: String, //来访时间 request: String, //受访页面 status: String, //状态 var valid: Boolean = true //判断是否合法 ) extends Serializablenginx 日志示例50.116.27.194 - - [18/Sep/2013:07:11:29 +0000] "POST /wp-cron.php?doing_wp_cron=1379488288.8893849849700927734375 HTTP/1.0" 200 0 "-" "WordPress/3.6; http://itunic.com"统计结果示例2013-09-18 /wp-admin/admin-ajax.php 200 2013-09-18 /wp-cron.php 73 2013-09-18 /batch.manage.php 21 2013-09-18 /index.php 10 2013-09-18 /tag/waitoutputthreads/index.php 10 2013-09-19 /wp-admin/admin-ajax.php 120 2013-09-19 /wp-cron.php 24 2013-09-19 /index.php 13 2013-09-19 /register.php 9 2013-09-19 /wp-admin/post.php 4 2013-09-19 /wp-admin/async-upload.php 3

【本文地址】

公司简介

联系我们