大資料教程（14.2）網站資料分析(二)
上一篇文章介紹了'網站點選流資料分析專案業務背景'；本篇部落格博主將繼續分享網站分析的相關知識。
一、整體技術流程及架構
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();
@Override
protected 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');