2019独角兽企业重金招聘Python工程师标准>>> 
上一篇文章介绍了'网站点击流数据分析项目业务背景'；本篇博客博主将继续分享网站分析的相关知识。
一、整体技术流程及架构
1.1.数据处理流程
该项目是一个纯粹的数据分析项目，其整体流程基本上就是依据数据的处理流程进行，依此有以下几个大的步骤：
1) 数据采集
首先，通过页面嵌入JS代码的方式获取用户访问行为，并发送到web服务的后台记录日志
然后，将各服务器上生成的点击流日志通过实时或批量的方式汇聚到HDFS文件系统中
当然，一个综合分析系统，数据源可能不仅包含点击流数据，还有数据库中的业务数据（如用户信息、商品信 息、订单信息等）及对分析有益的外部数据。
2) 数据预处理
通过mapreduce程序对采集到的点击流数据进行预处理，比如清洗，格式整理，滤除脏数据等
3) 数据入库
将预处理之后的数据导入到HIVE仓库中相应的库和表中
4) 数据分析
项目的核心内容，即根据需求开发ETL分析语句，得出各种统计结果
5) 数据展现
将分析所得数据进行可视化
1.2.项目结构
由于本项目是一个纯粹数据分析项目，其整体结构亦跟分析流程匹配，并没有特别复杂的结构，如下图：
其中，需要强调的是：系统的数据分析不是一次性的，而是按照一定的时间频率反复计算，因而整个处理链条中的各个环节需要按照一定的先后依赖关系紧密衔接，即涉及到大量任务单元的管理调度，所以，项目中需要添加一个任务调度模块(此处可由之前讲的azkaban替代OOZIE)。
1.3.数据展现
数据展现的目的是将分析所得的数据进行可视化，以便运营决策人员能更方便地获取数据，更快更简单地理解数据。可使用springmvc+echarts的简单架构实现。
二、模块开发——数据采集
2.1.需求
数据采集的需求广义上来说分为两大部分。
1）是在页面采集用户的访问行为，具体开发工作：
1、开发页面埋点js，采集用户访问行为
2、后台接受页面js请求记录日志
此部分工作也可以归属为“数据源”，其开发工作通常由web开发团队负责
2）是从web服务器上汇聚日志到HDFS，是数据分析系统的数据采集，此部分工作由数据分析平台建设团队负责，具体的技术实现有很多方式：
Shell脚本
优点：轻量级，开发简单
缺点：对日志采集过程中的容错处理不便控制
Java采集程序
优点：可对采集过程实现精细控制
缺点：开发工作量大
Flume日志采集框架
成熟的开源日志采集系统，且本身就是hadoop生态体系中的一员，与hadoop体系中的各种框架组件具有天生的亲和力，可扩展性强
2.2.技术选型
在点击流日志分析这种场景中，对数据采集部分的可靠性、容错能力要求通常不会非常严苛，因此使用通用的flume日志采集框架完全可以满足需求。
本项目即用flume来实现日志采集。
Flume日志采集系统搭建
1、数据源信息
本项目分析的数据用nginx服务器所生成的流量日志，存放在各台nginx服务器上，如：
/var/log/httpd/access_log.2015-11-10-13-00.log
/var/log/httpd/access_log.2015-11-10-14-00.log
/var/log/httpd/access_log.2015-11-10-15-00.log
/var/log/httpd/access_log.2015-11-10-16-00.log
2、数据内容样例
数据的具体内容在采集阶段其实不用太关心。
58.215.204.118 - - [18/Sep/2013:06:51:35 +0000] "GET /wp-includes/js/jquery/jquery.js?ver=1.10.2 HTTP/1.1" 304 0 "http://blog.fens.me/nodejs-socketio-chat/" "Mozilla/5.0 (Windows NT 5.1; rv:23.0) Gecko/20100101 Firefox/23.0" 字段解析：
1、访客ip地址： 58.215.204.118
2、访客用户信息： - -
3、请求时间：[18/Sep/2013:06:51:35 +0000]
4、请求方式：GET
5、请求的url：/wp-includes/js/jquery/jquery.js?ver=1.10.2
6、请求所用协议：HTTP/1.1
7、响应码：304
8、返回的数据流量：0
9、访客的来源url：http://blog.fens.me/nodejs-socketio-chat/
10、访客所用浏览器：Mozilla/5.0 (Windows NT 5.1; rv:23.0) Gecko/20100101 Firefox/23.0
3、日志文件生成规律
基本规律为：
当前正在写的文件为access_log；
文件体积达到256M，或时间间隔达到60分钟，即滚动重命名切换成历史日志文件；
形如： access_log.2015-11-10-13-00.log
当然，每个公司的web服务器日志策略不同，可在web程序的log4j.properties中定义，如下：
log4j.appender.logDailyFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.logDailyFile.layout = org.apache.log4j.PatternLayout
log4j.appender.logDailyFile.layout.ConversionPattern = [%-5p][%-22d{yyyy/MM/dd HH:mm:ssS}][%l]%n%m%n
log4j.appender.logDailyFile.Threshold = DEBUG
log4j.appender.logDailyFile.ImmediateFlush = TRUE
log4j.appender.logDailyFile.Append = TRUE
log4j.appender.logDailyFile.File = /var/logs/access_log
log4j.appender.logDailyFile.DatePattern = '.'yyyy-MM-dd-HH-mm'.log'
log4j.appender.logDailyFile.Encoding = UTF-84、Flume采集实现
Flume采集系统的搭建相对简单：
1、在个web服务器上部署agent节点，修改配置文件
2、启动agent节点，将采集到的数据汇聚到指定的HDFS目录中
如下图：
版本选择：apache-flume-1.7.0
采集规则设计：
1、采集源：nginx服务器日志目录
2、存放地：hdfs目录/home/hadoop/weblogs/
采集规则配置详情
agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1# Describe/configure spooldir source1
#agent1.sources.source1.type = spooldir
#agent1.sources.source1.spoolDir = /var/logs/nginx/
#agent1.sources.source1.fileHeader = false# Describe/configure tail -F source1
#使用exec作为数据源source组件
agent1.sources.source1.type = exec
#使用tail -F命令实时收集新产生的日志数据
agent1.sources.source1.command = tail -F /var/logs/nginx/access_log
agent1.sources.source1.channels = channel1#configure host for source
#配置一个拦截器插件
agent1.sources.source1.interceptors = i1
agent1.sources.source1.interceptors.i1.type = host
#使用拦截器插件获取agent所在服务器的主机名
agent1.sources.source1.interceptors.i1.hostHeader = hostname#配置sink组件为hdfs
agent1.sinks.sink1.type = hdfs
#a1.sinks.k1.channel = c1#agent1.sinks.sink1.hdfs.path=hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H%M%S
#指定文件sink到hdfs上的路径
agent1.sinks.sink1.hdfs.path=
hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M_%hostname
#指定文件名前缀
agent1.sinks.sink1.hdfs.filePrefix = access_log
agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
#指定每批下沉数据的记录条数
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text
#指定下沉文件按1G大小滚动
agent1.sinks.sink1.hdfs.rollSize = 1024*1024*1024
#指定下沉文件按1000000条数滚动
agent1.sinks.sink1.hdfs.rollCount = 1000000
#指定下沉文件按30分钟滚动
agent1.sinks.sink1.hdfs.rollInterval = 30
#agent1.sinks.sink1.hdfs.round = true
#agent1.sinks.sink1.hdfs.roundValue = 10
#agent1.sinks.sink1.hdfs.roundUnit = minute
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true# Use a channel which buffers events in memory
#使用memory类型channel
agent1.channels.channel1.type = memory
agent1.channels.channel1.keep-alive = 120
agent1.channels.channel1.capacity = 500000
agent1.channels.channel1.transactionCapacity = 600# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1
启动采集
在部署了flume的nginx服务器上，启动flume的agent，命令如下：
bin/flume-ng agent --conf ./conf -f ./conf/weblog.properties.2 -n agent
注意：启动命令中的 -n 参数要给配置文件中配置的agent名称
三、模块开发——数据预处理
3.1主要目的：a.过滤“不合规”数据；b.格式转换和规整；c.根据后续的统计需求，过滤分离出各种不同主题(不同栏目path)的基础数据
3.2实现方式：开发一个mr程序WeblogPreProcess(内容太长，见大数据hadoop之大型互联网电商公司网站日志分析、web日志数据清洗)
public class WeblogPreProcess {static class WeblogPreProcessMapper extends Mapper<LongWritable, Text, Text, NullWritable> {Text k = new Text();NullWritable v = NullWritable.get();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();WebLogBean webLogBean = WebLogParser.parser(line);
// WebLogBean productWebLog = WebLogParser.parser2(line);
// WebLogBean bbsWebLog = WebLogParser.parser3(line);
// WebLogBean cuxiaoBean = WebLogParser.parser4(line);if (!webLogBean.isValid())return;k.set(webLogBean.toString());context.write(k, v);
// k.set(productWebLog);
// context.write(k, v);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(WeblogPreProcess.class);job.setMapperClass(WeblogPreProcessMapper.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}
}运行mr对数据进行预处理
hadoop jar weblog.jar cn.itcast.bigdata.hive.mr.WeblogPreProcess /weblog/input /weblog/preout3.3点击流模型数据梳理
由于大量的指标统计从点击流模型中更容易得出，所以在预处理阶段，可以使用mr程序来生成点击流模型的数据
3.3.1点击流模型pageviews表
Pageviews表模型数据生成
代码见工程
hadoop jar weblogpreprocess.jar \
cn.itcast.bigdata.hive.mr.ClickStreamThree \
/user/hive/warehouse/dw_click.db/test_ods_weblog_origin/datestr=2013-09-20/ /test-click/pageviews/表结构：
(表定义及数据导入见6.2节)
3.3.2 点击流模型visit信息表
注：“一次访问”=“N次连续请求”
直接从原始数据中用hql语法得出每个人的“次”访问信息比较困难，可先用mapreduce程序分析原始数据得出“次”信息数据，然后再用hql进行更多维度统计
用MR程序从pageviews数据中，梳理出每一次visit的起止时间、页面信息
代码见工程
hadoop jar weblogpreprocess.jar cn.itcast.bigdata.hive.mr.ClickStreamVisit /weblog/sessionout /weblog/visitout然后，在hive仓库中建点击流visit模型表
load data inpath '/weblog/visitout' into table click_stream_visit partition(datestr='2013-09-18');
