linux流式訪問日志,流式實時日志分析系統的實現原理

我們知道網站用戶訪問流量是不間斷的，基于網站的訪問日志，即 Web log 分析是典型的流式實時計算應用場景。比如百度統計，它可以做流量分析、來源分析、網站分析、轉化分析。另外還有特定場景分析，比如安全分析，用來識別 CC 攻擊、 SQL 注入分析、脫庫等。這個項目就簡單實現一個類似于百度分析的系統。

實驗原理

百度統計(tongji.baidu.com)是百度推出的一款免費的專業網站流量分析工具，能夠告訴用戶訪客是如何找到并瀏覽用戶的網站的，以及在網站上瀏覽了哪些頁面。這些信息可以幫助用戶改善訪客在其網站上的使用體驗，不斷提升網站的投資回報率。

百度統計提供了幾十種圖形化報告，包括：趨勢分析、來源分析、頁面分析、訪客分析、定制分析等多種統計分析服務。

這里我們參考百度統計的功能，基于 Spark Streaming 簡單實現一個分析系統，使之包括以下分析功能。流量分析。一段時間內用戶網站的流量變化趨勢，針對不同的 IP 對用戶網站的流量進行細分。常見指標是總 PV 和各 IP 的PV。

來源分析。各種搜索引擎來源給用戶網站帶來的流量情況，需要精確到具體搜索引擎、具體關鍵詞。通過來源分析，用戶可以及時了解哪種類型的來源為其帶來了更多訪客。常見指標是搜索引擎、關鍵詞和終端類型的 PV 。

網站分析。各個頁面的訪問情況，包括及時了解哪些頁面最吸引訪客以及哪些頁面最容易導致訪客流失，從而幫助用戶更有針對性地改善網站質量。常見指標是各頁面的 PV 。

1 日志實時采集

Web log 一般在 HTTP 服務器收集，比如 Nginx access 日志文件。一個典型的方案是 Nginx 日志文件 + Flume + Kafka + Spark Streaming，如下所述：接收服務器用 Nginx ，根據負載可以部署多臺，數據落地至本地日志文件；

每個 Nginx 節點上部署 Flume ，使用 tail -f 實時讀取 Nginx 日志，發送至 KafKa 集群；

Spark Streaming 程序實時消費 Kafka 集群上的數據，實時分析，輸出；

結果寫入 MySQL 數據庫。

當然，還可以進一步優化，比如 CGI 程序直接發日志消息到 Kafka ，節省了寫訪問日志的磁盤開銷。這里主要專注 Spark Streaming 的應用，所以我們不做詳細論述。

2 流式分析系統實現

我們簡單模擬一下數據收集和發送的環節，用一個 Python 腳本隨機生成 Nginx 訪問日志，并通過腳本的方式自動上傳至 HDFS ，然后移動至指定目錄。 Spark Streaming 程序監控 HDFS 目錄，自動處理新的文件。

生成 Nginx 日志的 Python 代碼如下，保存為文件 sample_web_log.py 。#!/usr/bin/env?python#?-*-?coding:?utf-8?-*-import?random

import?timeclass?WebLogGeneration(object):

#?類屬性，由所有類的對象共享

site_url_base?=?"http://www.xxx.com/"

#?基本構造函數

def?__init__(self):????????#??前面7條是IE,所以大概瀏覽器類型70%為IE?，接入類型上，20%為移動設備，分別是7和8條,5%?為空

#??https://github.com/mssola/user_agent/blob/master/all_test.go

self.user_agent_dist?=?{0.0:"Mozilla/5.0?(compatible;?MSIE?10.0;?Windows?NT?6.2;?Trident/6.0)",????????????????????????????????0.1:"Mozilla/5.0?(compatible;?MSIE?10.0;?Windows?NT?6.2;?Trident/6.0)",????????????????????????????????0.2:"Mozilla/4.0?(compatible;?MSIE?7.0;?Windows?NT?5.1;?Trident/4.0;?.NET?CLR?2.0.50727)",????????????????????????????????0.3:"Mozilla/4.0?(compatible;?MSIE6.0;?Windows?NT?5.0;?.NET?CLR?1.1.4322)",????????????????????????????????0.4:"Mozilla/5.0?(Windows?NT?6.1;?Trident/7.0;?rv:11.0)?like?Gecko",????????????????????????????????0.5:"Mozilla/5.0?(Windows?NT?6.1;?WOW64;?rv:41.0)?Gecko/20100101?Firefox/41.0",????????????????????????????????0.6:"Mozilla/4.0?(compatible;?MSIE6.0;?Windows?NT?5.0;?.NET?CLR?1.1.4322)",????????????????????????????????0.7:"Mozilla/5.0?(iPhone;?CPU?iPhone?OS?7_0_3?like?Mac?OS?X)?AppleWebKit/537.51.1?(KHTML,?like?Gecko)?Version/7.0?Mobile/11B511?Safari/9537.53",????????????????????????????????0.8:"Mozilla/5.0?(Linux;?Android?4.2.1;?Galaxy?Nexus?Build/JOP40D)?AppleWebKit/535.19?(KHTML,?like?Gecko)?Chrome/18.0.1025.166?Mobile?Safari/535.19",????????????????????????????????0.9:"Mozilla/5.0?(Macintosh;?Intel?Mac?OS?X?10_10_5)?AppleWebKit/537.36?(KHTML,?like?Gecko)?Chrome/45.0.2454.85?Safari/537.36",????????????????????????????????1:"?",}????????self.ip_slice_list?=?[10,?29,?30,?46,?55,?63,?72,?87,?98,132,156,124,167,143,187,168,190,201,202,214,215,222]????????self.url_path_list?=?["login.php","view.php","list.php","upload.php","admin/login.php","edit.php","index.html"]????????self.http_refer?=?[?"http://www.baidu.com/s?wd={query}","http://www.google.cn/search?q={query}","http://www.sogou.com/web?query={query}","http://one.cn.yahoo.com/s?p={query}","http://cn.bing.com/search?q={query}"]????????self.search_keyword?=?["spark","hadoop","hive","spark?mlib","spark?sql"]????def?sample_ip(self):

slice?=?random.sample(self.ip_slice_list,?4)?#從ip_slice_list中隨機獲取4個元素，作為一個片斷返回

return??".".join([str(item)?for?item?in?slice])??#??todo

def?sample_url(self):????????return??random.sample(self.url_path_list,1)[0]????def?sample_user_agent(self):

dist_uppon?=?random.uniform(0,?1)????????return?self.user_agent_dist[float('%0.1f'?%?dist_uppon)]????#?主要搜索引擎referrer參數

def?sample_refer(self):????????if?random.uniform(0,?1)?>?0.2:??#?只有20%?流量有refer

return?"-"

refer_str=random.sample(self.http_refer,1)

query_str=random.sample(self.search_keyword,1)????????return?refer_str[0].format(query=query_str[0])????def?sample_one_log(self,count?=?3):

time_str?=?time.strftime("%Y-%m-%d?%H:%M:%S",time.localtime())????????while?count?>1:

query_log?=?"{ip}?-?-?[{local_time}]?\"GET?/{url}?HTTP/1.1\"?200?0?\"{refer}\"?\"{user_agent}\"?\"-\"".format(ip=self.sample_ip(),local_time=time_str,url=self.sample_url(),refer=self.sample_refer(),user_agent=self.sample_user_agent())

print?query_log

count?=?count?-1if?__name__?==?"__main__":

web_log_gene?=?WebLogGeneration()????#while?True:

#????time.sleep(random.uniform(0,?3))

web_log_gene.sample_one_log(random.uniform(10,?100))

這是一條日志的示例，為一行形式，各字段間用空格分隔，字符串類型的值用雙引號包圍：46.202.124.63?-?-?[2015-11-26?09:54:27]?"GET?/view.php?HTTP/1.1"?200?0?"http://www.google.cn/search?q=hadoop"?"Mozilla/5.0?(compatible;?MSIE?10.0;?Windows?NT?6.2;?Trident/6.0)"?"-"

然后需要一個簡單的腳本來調用上面的腳本以隨機生成日志，上傳至 HDFS ，然后移動到目標目錄：#!/bin/bash

#?HDFS命令?HDFS="/usr/local/myhadoop/hadoop-2.6.0/bin/hadoop?fs"#?Streaming程序監聽的目錄，注意跟后面Streaming程序的配置要保持一致?streaming_dir=”/spark/streaming”

#?清空舊數據?$HDFS?-rm?"${streaming_dir}"'/tmp/*'?>?/dev/null?2>&1

$HDFS?-rm?"${streaming_dir}"'/*'?????>?/dev/null?2>&1

#?一直運行?while?[?1?];?do

./sample_web_log.py?>?test.log

#?給日志文件加上時間戳，避免重名

tmplog="access.`date?+'%s'`.log"

#?先放在臨時目錄，再move至Streaming程序監控的目錄下，確保原子性

#?臨時目錄用的是監控目錄的子目錄，因為子目錄不會被監控

$HDFS?-put?test.log?${streaming_dir}/tmp/$tmplog

$HDFS?-mv???????????${streaming_dir}/tmp/$tmplog?${streaming_dir}/

echo?"`date?+"%F?%T"`?put?$tmplog?to?HDFS?succeed"

sleep?1done

Spark Streaming 程序代碼如下所示，可以在 bin/spark-shell 交互式環境下運行，如果要以 Spark 程序的方式運行，按注釋中的說明調整一下 StreamingContext 的生成方式即可。啟動 bin/spark-shell 時，為了避免因 DEBUG 日志信息太多而影響觀察輸出，可以將 DEBUG 日志重定向至文件，屏幕上只顯示主要輸出，方法是./bin/spark-shell 2>spark-shell-debug.log：//?導入類import?org.apache.spark.SparkConfimport?org.apache.spark.streaming.{Seconds,?StreamingContext}//?設計計算的周期，單位秒val?batch?=?10/*

*?這是bin/spark-shell交互式模式下創建StreamingContext的方法

*?非交互式請使用下面的方法來創建

*/val?ssc?=?new?StreamingContext(sc,?Seconds(batch))/*

//?非交互式下創建StreamingContext的方法

val?conf?=?new?SparkConf().setAppName("NginxAnay")

val?ssc?=?new?StreamingContext(conf,?Seconds(batch))

*//*

*?創建輸入DStream，是文本文件目錄類型

*?本地模式下也可以使用本地文件系統的目錄，比如?file:///home/spark/streaming

*/val?lines?=?ssc.textFileStream("hdfs:///spark/streaming")/*

*?下面是統計各項指標，調試時可以只進行部分統計，方便觀察結果

*///?1.?總PVlines.count().print()//?2.?各IP的PV，按PV倒序//???空格分隔的第一個字段就是IPlines.map(line?=>?{(line.split("?")(0),?1)}).reduceByKey(_?+?_).transform(rdd?=>?{

rdd.map(ip_pv?=>?(ip_pv._2,?ip_pv._1)).

sortByKey(false).

map(ip_pv?=>?(ip_pv._2,?ip_pv._1))

}).print()//?3.?搜索引擎PVval?refer?=?lines.map(_.split("\"")(3))//?先輸出搜索引擎和查詢關鍵詞，避免統計搜索關鍵詞時重復計算//?輸出(host,?query_keys)val?searchEnginInfo?=?refer.map(r?=>?{

val?f?=?r.split('/')

val?searchEngines?=?Map(????????"www.google.cn"?->?"q",????????"www.yahoo.com"?->?"p",????????"cn.bing.com"?->?"q",????????"www.baidu.com"?->?"wd",????????"www.sogou.com"?->?"query"

)????if?(f.length?>?2)?{

val?host?=?f(2)????????if?(searchEngines.contains(host))?{

val?query?=?r.split('?')(1)????????????if?(query.length?>?0)?{

val?arr_search_q?=?query.split('&').filter(_.indexOf(searchEngines(host)+"=")?==?0)????????????????if?(arr_search_q.length?>?0)

(host,?arr_search_q(0).split('=')(1))????????????????else

(host,?"")

}?else?{

(host,?"")

}

}?else

("",?"")

}?else

("",?"")

})//?輸出搜索引擎PVsearchEnginInfo.filter(_._1.length?>?0).map(p?=>?{(p._1,?1)}).reduceByKey(_?+?_).print()//?4.?關鍵詞PVsearchEnginInfo.filter(_._2.length?>?0).map(p?=>?{(p._2,?1)}).reduceByKey(_?+?_).print()//?5.?終端類型PVlines.map(_.split("\"")(5)).map(agent?=>?{

val?types?=?Seq("iPhone",?"Android")????var?r?=?"Default"

for?(t?

r?=?t

}

(r,?1)

}).reduceByKey(_?+?_).print()//?6.?各頁面PVlines.map(line?=>?{(line.split("\"")(1).split("?")(1),?1)}).reduceByKey(_?+?_).print()//?啟動計算,等待執行結束(出錯或Ctrl-C退出)ssc.start()

ssc.awaitTermination()

打開兩個終端，一個調用上面的 bash 腳本模擬提交日志，一個在交互式環境下運行上面的 Streaming 程序。你可以看到各項指標的輸出，比如某個批次下的輸出為(依次對應上面的 6 個計算項)：

總PV-------------------------------------------

Time:?1448533850000?ms

-------------------------------------------

44374

各IP的PV，按PV倒序-------------------------------------------?Time:?1448533850000?ms

-------------------------------------------?(72.63.87.30,30)

(63.72.46.55,30)

(98.30.63.10,29)

(72.55.63.46,29)

(63.29.10.30,29)

(29.30.63.46,29)

(55.10.98.87,27)

(46.29.98.30,27)

(72.46.63.30,27)

(87.29.55.10,26)

搜索引擎PV-------------------------------------------?Time:?1448533850000?ms?-------------------------------------------?(cn.bing.com,1745)

(www.baidu.com,1773)

(www.google.cn,1793)

(www.sogou.com,1845)

關鍵詞PV-------------------------------------------

Time:?1448533850000?ms

-------------------------------------------

(spark,1426)

(hadoop,1455)

(spark?sql,1429)

(spark?mlib,1426)

(hive,1420)

終端類型PV-------------------------------------------

Time:?1448533850000?ms

-------------------------------------------

(Android,4281)

(Default,35745)

(iPhone,4348)

各頁面PV-------------------------------------------

Time:?1448533850000?ms

-------------------------------------------

(/edit.php,6435)

(/admin/login.php,6271)

(/login.php,6320)

(/upload.php,6278)

(/list.php,6411)

(/index.html,6309)

(/view.php,6350)

查看數據更直觀的做法是用圖形來展示，常見做法是將結果寫入外部 DB ，然后通過一些圖形化報表展示系統展示出來。比如對于終端類型，我們可以用餅圖展示。

對于連續的數據，我們也可以用拆線圖來展示趨勢。比如某頁面的PV。

除了常規的每個固定周期進行一次統計，我們還可以對連續多個周期的數據進行統計。以統計總 PV 為例，上面的示例是每 10 秒統計一次，可能還需要每分鐘統計一次，相當于 6 個 10 秒的周期。我們可以利用窗口方法實現，不同的代碼如下：//?窗口方法必須配置checkpint，可以這樣配置：?ssc.checkpoint("hdfs:///spark/checkpoint")

//?這是常規每10秒一個周期的PV統計?lines.count().print()

//?這是每分鐘(連續多個周期)一次的PV統計?lines.countByWindow(Seconds(batch*6),?Seconds(batch*6)).print()

使用相同的辦法運行程序之后，我們首先會看到連續 6 次 10 秒周期的 PV 統計輸出：-------------------------------------------

Time:?1448535090000?ms

-------------------------------------------

1101

-------------------------------------------

Time:?1448535100000?ms

-------------------------------------------

816

-------------------------------------------

Time:?1448535110000?ms