概述

大數(shù)據(jù)時(shí)代，隨著數(shù)據(jù)量不斷增長(zhǎng)，存儲(chǔ)與計(jì)算集群的規(guī)模也逐漸擴(kuò)大，幾百上千臺(tái)的云計(jì)算環(huán)境已不鮮見。現(xiàn)在的集群所需要解決的問題不僅僅是高性能、高可靠性、高可擴(kuò)展性，還需要面對(duì)易維護(hù)性以及數(shù)據(jù)平臺(tái)內(nèi)部的數(shù)據(jù)共享性等諸多挑戰(zhàn)。優(yōu)秀的系統(tǒng)運(yùn)維平臺(tái)既能實(shí)現(xiàn)數(shù)據(jù)平臺(tái)各組件的集中式管理、方便系統(tǒng)運(yùn)維人員日常監(jiān)測(cè)、提升運(yùn)維效率，又能反饋系統(tǒng)運(yùn)行狀態(tài)給系統(tǒng)開發(fā)人員。例如采集數(shù)據(jù)倉庫的日志可以按照時(shí)間序列查看各數(shù)據(jù)庫實(shí)例各種級(jí)別的日志數(shù)量與占比，采集 DB2 表空間數(shù)據(jù)分析可得到數(shù)據(jù)庫集群健康狀態(tài)，分析應(yīng)用服務(wù)器的日志可以查看出錯(cuò)最多的模塊、下載最多的文件、使用最多的功能等。大數(shù)據(jù)時(shí)代的業(yè)務(wù)與運(yùn)維將緊密的結(jié)合在一起。

日志

1. 什么是日志

日志是帶時(shí)間戳的基于時(shí)間序列的機(jī)器數(shù)據(jù)，包括 IT 系統(tǒng)信息（服務(wù)器、網(wǎng)絡(luò)設(shè)備、操作系統(tǒng)、應(yīng)用軟件）、物聯(lián)網(wǎng)各種傳感器信息。日志可以反映用戶實(shí)際行為，是真實(shí)的數(shù)據(jù)。

2. 日志處理方案演進(jìn)

圖 1. 日志處理方案經(jīng)歷的版本迭代

• 日志處理 v1.0：日志沒有集中式處理；只做事后追查，黑客入侵后刪除日志無法察覺；使用數(shù)據(jù)庫存儲(chǔ)日志，無法勝任復(fù)雜事務(wù)處理。

• 日志處理 v2.0：使用 Hadoop 平臺(tái)實(shí)現(xiàn)日志離線批處理，缺點(diǎn)是實(shí)時(shí)性差；使用 Storm 流處理框架、Spark 內(nèi)存計(jì)算框架處理日志，但 Hadoop/Storm/Spark 都是編程框架，并不是拿來即用的平臺(tái)。

• 日志處理 v3.0：使用日志實(shí)時(shí)搜索引擎分析日志，特點(diǎn)：第一是快，日志從產(chǎn)生到搜索分析出結(jié)果只有數(shù)秒延時(shí)；第二是大，每天處理 TB 日志量；第三是靈活，可搜索分析任何日志。作為代表的解決方案有 Splunk、ELK、SILK。

圖 2. 深度整合 ELK、Spark、Hadoop 構(gòu)建日志分析系統(tǒng)

ELK Stack

ELK Stack 是開源日志處理平臺(tái)解決方案，背后的商業(yè)公司是 Elastic(https://www.elastic.co/)。它由日志采集解析工具 Logstash、基于 Lucene 的全文搜索引擎 Elasticsearch、分析可視化平臺(tái) Kibana 組成。目前 ELK 的用戶有 Adobe、Microsoft、Mozilla、Facebook、Stackoverflow、Cisco、ebay、Uber 等諸多知名廠商。

1. Logstash

Logstash 是一種功能強(qiáng)大的信息采集工具，類似于 Hadoop 生態(tài)圈里的 Flume。通常在其配置文件規(guī)定 Logstash 如何處理各種類型的事件流，一般包含 input、filter、output 三個(gè)部分。Logstash 為各個(gè)部分提供相應(yīng)的插件，因而有 input、filter、output 三類插件完成各種處理和轉(zhuǎn)換；另外 codec 類的插件可以放在 input 和 output 部分通過簡(jiǎn)單編碼來簡(jiǎn)化處理過程。下面以 DB2 的一條日志為例。

圖 3.DB2 數(shù)據(jù)庫產(chǎn)生的半結(jié)構(gòu)化日志樣例

這是一種多行的日志，每一條日志以：“2014-10-19-12.19.46.033037-300”格式的時(shí)間戳為起始標(biāo)志。可以在 input 部分引入 codec 插件 multiline，來將一條日志的多行文本封裝到一條消息（message）中。

1 2 3 4 5 6 7 8 9 10

input { ?file { ?path => "path/to/filename" ?codec => multiline { ?pattern => "^\d{4}-\d{2}-\d{2}-\d{2}\.\d{2}\.\d{2}\.\d{6}[\+-]\d{3}" ?negate => true ?what => previous ?} ?} }

使用 file 插件導(dǎo)入文件形式的日志，而嵌入的 codec 插件 multiline 的參數(shù) pattern 就規(guī)定了分割日志條目的時(shí)間戳格式。在 DataStage 多行日志的實(shí)際應(yīng)用中，有時(shí)一條日志會(huì)超過 500 行，這超出了 multiline 組件默認(rèn)的事件封裝的最大行數(shù)，這需要我們?cè)?multiline 中設(shè)置 max_lines 屬性。

經(jīng)過 input 部分讀入預(yù)處理后的數(shù)據(jù)流入 filter 部分，其使用 grok、mutate 等插件來過濾文本和匹配字段，并且我們自己可以為事件流添加額外的字段信息：

1 2 3 4 5 6 7 8 9 10 11 12 13

filter { ?mutate{ ?gsub => ['message', "\n", " "] ?} ?grok { ?match => { "message" => ?"(?<timestamp>%{YEAR}-%{MONTHNUM}-%{MONTHDAY}-%{HOUR}\.%{MINUTE}\.%{SECOND})%{INT:timezone}(?:%{SPACE}%{WORD:recordid}%{SPACE})(?:LEVEL%{SPACE}:%{SPACE}%{DATA:level}%{SPACE})(?:PID%{SPACE}:%{SPACE}%{INT:processid}%{SPACE})(?:TID%{SPACE}:%{SPACE}%{INT:threadid}%{SPACE})(?:PROC%{SPACE}:%{SPACE}%{DATA:process}%{SPACE})?(?:INSTANCE%{SPACE}:%{SPACE}%{WORD:instance}%{SPACE})?(?:NODE%{SPACE}:%{SPACE}%{WORD:node}%{SPACE})?(?:DB%{SPACE}:%{SPACE}%{WORD:dbname}%{SPACE})?(?:APPHDL%{SPACE}:%{SPACE}%{NOTSPACE:apphdl}%{SPACE})?(?:APPID%{SPACE}:%{SPACE}%{NOTSPACE:appid}%{SPACE})?(?:AUTHID%{SPACE}:%{SPACE}%{WORD:authid}%{SPACE})?(?:HOSTNAME%{SPACE}:%{SPACE}%{HOSTNAME:hostname}%{SPACE})?(?:EDUID%{SPACE}:%{SPACE}%{INT:eduid}%{SPACE})?(?:EDUNAME%{SPACE}:%{SPACE}%{DATA:eduname}%{SPACE})?(?:FUNCTION%{SPACE}:%{SPACE}%{DATA:function}%{SPACE})(?:probe:%{SPACE}%{INT:probe}%{SPACE})%{GREEDYDATA:functionlog}" ?} ?} ?date { ?match => [ "timestamp", "YYYY-MM-dd-HH.mm.ss.SSSSSS" ] ?} ?}

前面 input 部分的 multiline 插件將一條多行日志項(xiàng)轉(zhuǎn)化為一行，并以“\n”替代實(shí)際的換行符。為了便于后面處理，這里的 mutate 插件就是將這些“\n”替換為空格。而 grok 插件用于匹配提取日志項(xiàng)中有意義的字段信息。最后的 date 插件則是按格式“YYYY-MM-dd-HH.mm.ss.SSSSSS”解析提取的時(shí)間戳字段，并賦給系統(tǒng)默認(rèn)的時(shí)間戳字段“@timestamp”。Output 插件用于指定事件流的去向，可以是消息隊(duì)列、全文搜索引擎、TCP Socket、Email 等幾十種目標(biāo)端。

2. Elasticsearch

Elasticsearch 是基于 Lucene 的近實(shí)時(shí)搜索平臺(tái)，它能在一秒內(nèi)返回你要查找的且已經(jīng)在 Elasticsearch 做了索引的文檔。它默認(rèn)基于 Gossip 路由算法的自動(dòng)發(fā)現(xiàn)機(jī)制構(gòu)建配置有相同 cluster name 的集群，但是有的時(shí)候這種機(jī)制并不可靠，會(huì)發(fā)生腦裂現(xiàn)象。鑒于主動(dòng)發(fā)現(xiàn)機(jī)制的不穩(wěn)定性，用戶可以選擇在每一個(gè)節(jié)點(diǎn)上配置集群其他節(jié)點(diǎn)的主機(jī)名，在啟動(dòng)集群時(shí)進(jìn)行被動(dòng)發(fā)現(xiàn)。

Elasticsearch 中的 Index 是一組具有相似特征的文檔集合，類似于關(guān)系數(shù)據(jù)庫模型中的數(shù)據(jù)庫實(shí)例，Index 中可以指定 Type 區(qū)分不同的文檔，類似于數(shù)據(jù)庫實(shí)例中的關(guān)系表，Document 是存儲(chǔ)的基本單位，都是 JSON 格式，類似于關(guān)系表中行級(jí)對(duì)象。我們處理后的 JSON 文檔格式的日志都要在 Elasticsearch 中做索引，相應(yīng)的 Logstash 有 Elasticsearch output 插件，對(duì)于用戶是透明的。

Hadoop 生態(tài)圈為大規(guī)模數(shù)據(jù)集的處理提供多種分析功能，但實(shí)時(shí)搜索一直是 Hadoop 的軟肋。如今，Elasticsearch for Apache Hadoop（ES-Hadoop）彌補(bǔ)了這一缺陷，為用戶整合了 Hadoop 的大數(shù)據(jù)分析能力以及 Elasticsearch 的實(shí)時(shí)搜索能力.

圖 4. 應(yīng)用 es-hadoop 整合 Hadoop Ecosystem 與 Elasticsearch 架構(gòu)圖（https://www.elastic.co/products/hadoop）

3. Kibana

Kibana 是專門設(shè)計(jì)用來與 Elasticsearch 協(xié)作的，可以自定義多種表格、柱狀圖、餅狀圖、折線圖對(duì)存儲(chǔ)在 Elasticsearch 中的數(shù)據(jù)進(jìn)行深入挖掘分析與可視化。下圖定制的儀表盤可以動(dòng)態(tài)監(jiān)測(cè)數(shù)據(jù)庫集群中每個(gè)數(shù)據(jù)庫實(shí)例產(chǎn)生的各種級(jí)別的日志。

圖 5. 實(shí)時(shí)監(jiān)測(cè) DB2 實(shí)例運(yùn)行狀態(tài)的動(dòng)態(tài)儀表盤

Kafka

Kafka 是 LinkedIn 開源的分布式消息隊(duì)列，它采用了獨(dú)特的消費(fèi)者-生產(chǎn)者架構(gòu)實(shí)現(xiàn)數(shù)據(jù)平臺(tái)各組件間的數(shù)據(jù)共享。集群概念中的 server 在 Kafka 中稱之為 broker，它使用主題管理不同類別的數(shù)據(jù)，比如 DB2 日志歸為一個(gè)主題，tomcat 日志歸為一個(gè)主題。我們使用 Logstash 作為 Kafka 消息的生產(chǎn)者時(shí)，output 插件就需要配置好 Kafka broker 的列表，也就是 Kafka 集群主機(jī)的列表；相應(yīng)的，用作 Kafka 消費(fèi)者角色的 Logstash 的 input 插件就要配置好需要訂閱的 Kafka 中的主題名稱和 ZooKeeper 主機(jī)列表。Kafka 通過將數(shù)據(jù)持久化到硬盤的 Write Ahead Log（WAL）保證數(shù)據(jù)可靠性與順序性，但這并不會(huì)影響實(shí)時(shí)數(shù)據(jù)的傳輸速度，實(shí)時(shí)數(shù)據(jù)仍是通過內(nèi)存?zhèn)鬏數(shù)摹afka 是依賴于 ZooKeeper 的，它將每組消費(fèi)者消費(fèi)的相應(yīng) topic 的偏移量保存在 ZooKeeper 中。據(jù)稱 LinkedIn 內(nèi)部的 Kafka 集群每天已能處理超過 1 萬億條消息。

圖 6. 基于消息訂閱機(jī)制的 Kafka 架構(gòu)

除了可靠性和獨(dú)特的 push&pull 架構(gòu)外，相較于其他消息隊(duì)列，Kafka 還擁有更大的吞吐量：

圖 7. 基于消息持久化機(jī)制的消息隊(duì)列吞吐量比較

Spark Streaming

Spark 由加州大學(xué)伯克利分校 AMP 實(shí)驗(yàn)室 (Algorithms, Machines, and People Lab) 開發(fā)，可用來構(gòu)建大型的、低延遲的數(shù)據(jù)分析應(yīng)用程序。它將批處理、流處理、即席查詢?nèi)跒橐惑w。Spark 社區(qū)也是相當(dāng)火爆，平均每三個(gè)月迭代一次版本更是體現(xiàn)了它在大數(shù)據(jù)處理領(lǐng)域的地位。

Spark Streaming 不同于 Storm，Storm 是基于事件級(jí)別的流處理，Spark Streaming 是 mini-batch 形式的近似流處理的微型批處理。Spark Streaming 提供了兩種從 Kafka 中獲取消息的方式：

第一種是利用 Kafka 消費(fèi)者高級(jí) API 在 Spark 的工作節(jié)點(diǎn)上創(chuàng)建消費(fèi)者線程，訂閱 Kafka 中的消息，數(shù)據(jù)會(huì)傳輸?shù)?Spark 工作節(jié)點(diǎn)的執(zhí)行器中，但是默認(rèn)配置下這種方法在 Spark Job 出錯(cuò)時(shí)會(huì)導(dǎo)致數(shù)據(jù)丟失，如果要保證數(shù)據(jù)可靠性，需要在 Spark Streaming 中開啟Write Ahead Logs（WAL），也就是上文提到的 Kafka 用來保證數(shù)據(jù)可靠性和一致性的數(shù)據(jù)保存方式。可以選擇讓 Spark 程序把 WAL 保存在分布式文件系統(tǒng)（比如 HDFS）中。

第二種方式不需要建立消費(fèi)者線程，使用 createDirectStream 接口直接去讀取 Kafka 的 WAL，將 Kafka 分區(qū)與 RDD 分區(qū)做一對(duì)一映射，相較于第一種方法，不需再維護(hù)一份 WAL 數(shù)據(jù)，提高了性能。讀取數(shù)據(jù)的偏移量由 Spark Streaming 程序通過檢查點(diǎn)機(jī)制自身處理，避免在程序出錯(cuò)的情況下重現(xiàn)第一種方法重復(fù)讀取數(shù)據(jù)的情況，消除了 Spark Streaming 與 ZooKeeper/Kafka 數(shù)據(jù)不一致的風(fēng)險(xiǎn)。保證每條消息只會(huì)被 Spark Streaming 處理一次。以下代碼片通過第二種方式讀取 Kafka 中的數(shù)據(jù)：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

// Create direct kafka stream with brokers and topics JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream( ?jssc, ?String.class, ?String.class, ?StringDecoder.class, ?StringDecoder.class, ?kafkaParams, ?topicsSet); messages.foreachRDD(new Function<JavaPairRDD<String,String>,Void>(){ ????public Void call(JavaPairRDD<String, String> v1) ?throws Exception { ????????v1.foreach(new VoidFunction<Tuple2<String, String>>(){ ?public void call(Tuple2<String, String> tuple2) { ?try{ ?JSONObject a = new JSONObject(tuple2._2); ?...

Spark Streaming 獲取到消息后便可以通過 Tuple 對(duì)象自定義操作消息，如下圖是針對(duì) DB2 數(shù)據(jù)庫日志的郵件告警，生成告警郵件發(fā)送到 Notes 郵箱：

圖 8. 基于 Spark Streaming 對(duì) DB2 異常日志實(shí)現(xiàn) Notes 郵件告警

互聯(lián)網(wǎng)行業(yè)日志處理方案舉例介紹與應(yīng)用

1. 新浪

新浪采用的技術(shù)架構(gòu)是常見的 Kafka 整合 ELK Stack 方案。Kafka 作為消息隊(duì)列用來緩存用戶日志；使用 Logstash 做日志解析，統(tǒng)一成 JSON 格式輸出給 Elasticsearch；使用 Elasticsearch 提供實(shí)時(shí)日志分析與強(qiáng)大的搜索和統(tǒng)計(jì)服務(wù)；Kibana 用作數(shù)據(jù)可視化組件。該技術(shù)架構(gòu)目前服務(wù)的用戶包括微博、微盤、云存儲(chǔ)、彈性計(jì)算平臺(tái)等十多個(gè)部門的多個(gè)產(chǎn)品的日志搜索分析業(yè)務(wù)，每天處理約 32 億條（2TB）日志。

新浪的日志處理平臺(tái)團(tuán)隊(duì)對(duì) Elasticsearch 做了大量?jī)?yōu)化（比如調(diào)整 max open files 等），并且開發(fā)了一個(gè)獨(dú)立的 Elasticsearch Index 管理系統(tǒng)，負(fù)責(zé)索引日常維護(hù)任務(wù)（比如索引的創(chuàng)建、優(yōu)化、刪除、與分布式文件系統(tǒng)的數(shù)據(jù)交換等）的調(diào)度及執(zhí)行。為 Elasticsearch 安裝了國內(nèi)中文分詞插件 elasticsearch-analysis-ik，滿足微盤搜索對(duì)中文分詞的需求。（見參考資源 2）

2. 騰訊

騰訊藍(lán)鯨數(shù)據(jù)平臺(tái)告警系統(tǒng)的技術(shù)架構(gòu)同樣基于分布式消息隊(duì)列和全文搜索引擎。但騰訊的告警平臺(tái)不僅限于此，它的復(fù)雜的指標(biāo)數(shù)據(jù)統(tǒng)計(jì)任務(wù)通過使用 Storm 自定義流式計(jì)算任務(wù)的方法實(shí)現(xiàn)，異常檢測(cè)的實(shí)現(xiàn)利用了曲線的時(shí)間周期性和相關(guān)曲線之間的相關(guān)性去定義動(dòng)態(tài)的閾值，并且基于機(jī)器學(xué)習(xí)算法實(shí)現(xiàn)了復(fù)雜的日志自動(dòng)分類（比如 summo logic）。

告警平臺(tái)把撥測(cè)（定時(shí) curl 一下某個(gè) url，有問題就告警）、日志集中檢索、日志告警（5 分鐘 Error 大于 X 次告警）、指標(biāo)告警（cpu 使用率大于 X 告警）整合進(jìn)同一個(gè)數(shù)據(jù)管線，簡(jiǎn)化了整體的架構(gòu)。（見參考資源 3）

3. 七牛

七牛采用的技術(shù)架構(gòu)為 Flume＋Kafka＋Spark，混部在 8 臺(tái)高配機(jī)器。根據(jù)七牛技術(shù)博客提供的數(shù)據(jù)，該日志處理平臺(tái)每天處理 500 億條數(shù)據(jù)，峰值 80 萬 TPS。?

Flume 相較于 Logstash 有更大的吞吐量，而且與 HDFS 整合的性能比 Logstash 強(qiáng)很多。七牛技術(shù)架構(gòu)選型顯然考慮了這一點(diǎn)，七牛云平臺(tái)的日志數(shù)據(jù)到 Kafka 后，一路同步到 HDFS，用于離線統(tǒng)計(jì)，另一路用于使用 Spark Streaming 進(jìn)行實(shí)時(shí)計(jì)算，計(jì)算結(jié)果保存在 Mongodb 集群中。（見參考資源 4）

任何解決方案都不是十全十美的，具體采用哪些技術(shù)要深入了解自己的應(yīng)用場(chǎng)景。就目前日志處理領(lǐng)域的開源組件來說，在以下幾個(gè)方面還比較欠缺：

• Logstash 的內(nèi)部狀態(tài)獲取不到，目前沒有好的成熟的監(jiān)控方案。
• Elasticsearch 具有海量存儲(chǔ)海量聚合的能力，但是同 Mongodb 一樣，并不適合于寫入數(shù)據(jù)非常多（1 萬 TPS 以上）的場(chǎng)景。
• 缺乏真正實(shí)用的異常檢測(cè)方法；實(shí)時(shí)統(tǒng)計(jì)方面缺乏成熟的解決方案，Storm 就是一個(gè)底層的執(zhí)行引擎，而 Spark 還缺少時(shí)間窗口等抽象。
• 對(duì)于日志自動(dòng)分類，還沒有開源工具可以做到 summo logic 那樣的效果。

結(jié)束語

大數(shù)據(jù)時(shí)代的運(yùn)維管理意義重大，好的日志處理平臺(tái)可以事半功倍的提升開發(fā)人員和運(yùn)維人員的效率。本文通過簡(jiǎn)單用例介紹了 ELK Stack、Kafka 和 Spark Streaming 在日志處理平臺(tái)中各自在系統(tǒng)架構(gòu)中的功能。現(xiàn)實(shí)中應(yīng)用場(chǎng)景繁多復(fù)雜、數(shù)據(jù)形式多種多樣，日志處理工作不是一蹴而就的，分析處理過程還需要在實(shí)踐中不斷挖掘和優(yōu)化，筆者也將致力于 DB2 數(shù)據(jù)庫運(yùn)行狀態(tài)更細(xì)節(jié)數(shù)據(jù)的收集和更全面細(xì)致的監(jiān)控。

相關(guān)主題

• 1. 參考“日志: 每個(gè)軟件工程師都應(yīng)該知曉的實(shí)時(shí)數(shù)據(jù)集成提取的那點(diǎn)事”，詳細(xì)了解日志處理。
• 2. 查看文章“新浪是如何分析處理 32 億條實(shí)時(shí)日志的？”，了解新浪的日志處理方案設(shè)計(jì)。
• 3. 查看文章“騰訊藍(lán)鯨數(shù)據(jù)平臺(tái)之告警系統(tǒng)”，了解騰訊在大型系統(tǒng)運(yùn)維工程上的實(shí)踐。
• 4. 查看文章“七牛是如何搞定每天 500 億條日志的”，了解七牛海量日志處理解決方案。
• 5. 查看文章“Mesos 在去哪兒網(wǎng)的應(yīng)用”，了解去哪兒網(wǎng)利用日志在集群管理中的實(shí)踐。
• 6. 參考網(wǎng)站“https://softwaremill.com/mqperf/”，了解基于消息持久化的各消息隊(duì)列各種負(fù)載下的性能分析。
• 在 developerWorks 大數(shù)據(jù)和分析專區(qū)，了解關(guān)于大數(shù)據(jù)的更多信息，獲取技術(shù)文檔、how-to 文章、培訓(xùn)、下載、產(chǎn)品信息以及其他資源。

https://www.ibm.com/developerworks/cn/analytics/library/ba-1512-elkstack-logprocessing/index.html

總結(jié)

以上是生活随笔為你收集整理的基于 ELK Stack 和 Spark Streaming 的日志处理平台设计与实现的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。