李柯

（西北工業(yè)大學(xué) 陜西省西安市 710000）

1 引言

數(shù)據(jù)的演變非常壯觀(guān)，在過(guò)去數(shù)據(jù)是字節(jié)級(jí)的，現(xiàn)在一些公司在同一時(shí)間使用大量PB 級(jí)數(shù)據(jù)。美國(guó)阿什維爾國(guó)家氣候數(shù)據(jù)中心的專(zhuān)家估計(jì)如果我們想存儲(chǔ)這個(gè)世界上所有的數(shù)據(jù)我們至少需要1200 個(gè)exabytes，但不可能確定相關(guān)號(hào)碼。因此，如何從海量數(shù)據(jù)中提取獲得有效的信息成為了大數(shù)據(jù)產(chǎn)業(yè)發(fā)展的首要解決問(wèn)題之一。

數(shù)據(jù)采集，越來(lái)越成為大數(shù)據(jù)產(chǎn)業(yè)發(fā)展的根基。數(shù)據(jù)采集作為從數(shù)據(jù)被測(cè)模塊和其他測(cè)量仿真模塊中采集信息的過(guò)程，經(jīng)歷多年的發(fā)展，在數(shù)據(jù)采集的過(guò)程中增添了數(shù)據(jù)的分類(lèi)和歸并操作，為后續(xù)的數(shù)據(jù)處理操作提供了便捷，并簡(jiǎn)化了相關(guān)后續(xù)操作。從圖1 中描述的Hadoop業(yè)務(wù)的開(kāi)發(fā)流程中，可以看到數(shù)據(jù)采集在Hadoop 開(kāi)發(fā)過(guò)程中作為十分重要的步驟，也是不可避免的步驟。

圖1： Hadoop 業(yè)務(wù)的開(kāi)發(fā)流程

在當(dāng)今的互聯(lián)網(wǎng)世界，日志文件的分析變得越來(lái)越重要。為了分析日志數(shù)據(jù)，我們利用網(wǎng)絡(luò)挖掘的幫助從這些日志數(shù)據(jù)中獲得所需的知識(shí)。而日志文件生成的速度極快，一天之內(nèi)，一個(gè)數(shù)據(jù)中心就能生成數(shù)十TB 的日志數(shù)據(jù)。傳統(tǒng)的數(shù)據(jù)采集數(shù)據(jù)來(lái)源于單一，且儲(chǔ)存、管理方式較為復(fù)雜，解析的數(shù)據(jù)量也相對(duì)較小，而現(xiàn)代大數(shù)據(jù)采集來(lái)源廣泛、數(shù)據(jù)量極大，數(shù)據(jù)類(lèi)型繁雜，需要分布式的數(shù)據(jù)庫(kù)存儲(chǔ)。顯然傳統(tǒng)的數(shù)據(jù)采集方式無(wú)法適應(yīng)現(xiàn)代大數(shù)據(jù)采集的需求，而現(xiàn)在很多互聯(lián)網(wǎng)企業(yè)所采用的事件日志采集方式，在應(yīng)用體系與數(shù)據(jù)分析體系間搭建了橋梁，從而將關(guān)聯(lián)解耦。在支持實(shí)時(shí)分析系統(tǒng)的同時(shí)也支持離線(xiàn)分析系統(tǒng)，并能在數(shù)據(jù)量增加的場(chǎng)合，通過(guò)增加節(jié)點(diǎn)數(shù)量水平拓展。

2 Flume簡(jiǎn)介

Flume作為一款實(shí)時(shí)日志采集系統(tǒng)，獲得了行業(yè)的肯定和應(yīng)用。從初始發(fā)行的Flume OG（original generation）到現(xiàn)在經(jīng)歷核心組件、核心配置以及代碼架構(gòu)重構(gòu)的Flume NG（next generation），flume 已經(jīng)是一款十分成熟的日志采集系統(tǒng)。本文也將著重研究Flume NG 的應(yīng)用改進(jìn)。

作為一款分布式、安全、高效率的實(shí)時(shí)日志采集系統(tǒng)，flume 將海量數(shù)據(jù)進(jìn)行集成，并可以自由更改數(shù)據(jù)發(fā)送方、對(duì)數(shù)據(jù)進(jìn)行簡(jiǎn)單處理交付數(shù)據(jù)接收分析方。

Flume 不僅提供了采集功能，還提供了對(duì)數(shù)據(jù)進(jìn)行簡(jiǎn)單處理并寫(xiě)入數(shù)據(jù)接收模塊的功能。在數(shù)據(jù)流動(dòng)的過(guò)程中，flume 以事件（event）作為數(shù)據(jù)載體或數(shù)據(jù)基本單位，事件攜帶日志數(shù)據(jù)并且在頭部加入標(biāo)識(shí)信息。一個(gè)flume 進(jìn)程被稱(chēng)為agent，agent 作為flume 的核心由source、channel、sink 組件構(gòu)成，這三個(gè)組件依次串行工作，而event 進(jìn)入flume 的方式就是通過(guò)被source 組件捕獲事件生成。Channel作為一個(gè)緩沖區(qū)，在sink 處理完一個(gè)事務(wù)之前保存事務(wù)數(shù)據(jù)，sink 將日志數(shù)據(jù)寫(xiě)入磁盤(pán)或者將事務(wù)傳入下一個(gè)Source。Flume NG 的體系結(jié)構(gòu)如圖2 和圖3 所示，這里特別強(qiáng)調(diào)的是channel 組件的重要性，由于sink 處理事件數(shù)據(jù)時(shí)有概率出錯(cuò)，導(dǎo)致數(shù)據(jù)丟失或出錯(cuò)，在sink 正確處理事件之前，需要channel 對(duì)事件進(jìn)行保存，以免事件處理出錯(cuò)后的數(shù)據(jù)丟失。

圖2： Flume 結(jié)構(gòu)示意

圖3： Flume 模型

3 Flume+Kafka

5G 時(shí)代的到來(lái)以及發(fā)展，極大地推動(dòng)了公共媒體的發(fā)展，加速了數(shù)據(jù)的流動(dòng)，數(shù)據(jù)量的極速增長(zhǎng)，各種設(shè)備記錄、消費(fèi)記錄、瀏覽記錄都成為了大數(shù)據(jù)的來(lái)源。由于每天有大量的數(shù)據(jù)流入流出，平臺(tái)產(chǎn)生大量的日志文件，文件中的數(shù)據(jù)包含了用戶(hù)喜好、輿論走向、社會(huì)習(xí)慣等信息。這些數(shù)據(jù)的數(shù)據(jù)源多種多樣，而數(shù)據(jù)本身體量龐大、變化迅速，如何保質(zhì)量、高效率、高可靠、避免重復(fù)的收集這些數(shù)據(jù)成為后續(xù)數(shù)據(jù)分析的前提之一。Flume 作為較近出現(xiàn)的實(shí)時(shí)日志采集系統(tǒng)，可以很好的利用自身的特點(diǎn)滿(mǎn)足公共媒體等行業(yè)數(shù)據(jù)采集的需求。本文將基于flume + Kafka 實(shí)時(shí)日志采集系統(tǒng)對(duì)數(shù)據(jù)收集平臺(tái)的搭建和優(yōu)化進(jìn)行研究，以滿(mǎn)足公共媒體等行業(yè)數(shù)據(jù)采集、數(shù)據(jù)分析的需求。

圖2 表示的體系結(jié)構(gòu)圖中，flume 的核心agent 前端連接服務(wù)器，后端連接Hadoop 分布式文件系統(tǒng)，這是通常的flume 結(jié)構(gòu)表示圖。然而使用時(shí)，通常將Flume 和Kafka 集成。那么為什么這樣做？先從業(yè)務(wù)需求出發(fā)，flume + Kafka 架構(gòu)通常都是為了進(jìn)行實(shí)時(shí)流式的日志處理，處理完成之后在后端連接流式實(shí)時(shí)數(shù)據(jù)處理工具，以便于達(dá)成日志實(shí)時(shí)分析的目的。如果在實(shí)時(shí)計(jì)算框架中直接使用flume，如果數(shù)據(jù)采集速率大于數(shù)據(jù)處理速率，就極易出現(xiàn)數(shù)據(jù)堆積甚至數(shù)據(jù)丟失，而Kafka 是一個(gè)可持久化的分布式消息緩存隊(duì)列，我們可以將其看成一個(gè)數(shù)據(jù)庫(kù)系統(tǒng)，能夠存儲(chǔ)一段時(shí)間的歷史數(shù)據(jù)，達(dá)到緩存數(shù)據(jù)的目的。

Kafka 作為一款開(kāi)源分布式事件流媒體平臺(tái)被成千上萬(wàn)家企業(yè)用于高性能數(shù)據(jù)管道、流媒體分析、大數(shù)據(jù)集成等領(lǐng)域。它能夠使生產(chǎn)集群的范圍擴(kuò)展到極大的代理規(guī)模、每天高速的消息流、數(shù)PB 的數(shù)據(jù)規(guī)模以及數(shù)以萬(wàn)計(jì)的分區(qū)，并彈性地?cái)U(kuò)展和收縮、存儲(chǔ)和管理。從而在可用性范圍上更高效地?cái)U(kuò)展整個(gè)集群，或者跨越地域范圍連接單獨(dú)的集群。

Kafka 作為中間件，一個(gè)突出的優(yōu)點(diǎn)就是將各層解耦，使得出錯(cuò)時(shí)由于其封裝性不會(huì)影響其他部分工作，并不影響其他功能。數(shù)據(jù)從數(shù)據(jù)源中被捕獲，到flume 中采集分類(lèi)，再到Kafka 中處理分類(lèi)時(shí)，數(shù)據(jù)一方面能夠同步到HDFS 進(jìn)行離線(xiàn)運(yùn)算，另一方面也能夠進(jìn)行實(shí)時(shí)運(yùn)算，實(shí)現(xiàn)數(shù)據(jù)的并發(fā)操作。同時(shí)Kafka 的開(kāi)箱即用式連接接口還整合了數(shù)百個(gè)事件源和事件接收器，使得Kafka 可以連接幾乎任何接口。圖4 表示了flume 與Kafka 搭配之后的結(jié)構(gòu)。

圖4： Kafka topic 數(shù)據(jù)收集以及flume 的sink 將數(shù)據(jù)傳入HDFS

3.1 Flume

Flume 系統(tǒng)的核心基礎(chǔ)單元agent，共包含以下三個(gè)組件：source、channel、sink。但是組件中的channel 如使用MemoryChannel，雖然可以實(shí)現(xiàn)高速傳輸?shù)溆邢薜娜萘吭跀?shù)據(jù)傳輸高峰時(shí)刻或agent 中止時(shí)會(huì)造成數(shù)據(jù)丟失；如使用FileChannel，有較大的容量在agent 中止后可以恢復(fù)數(shù)據(jù)，但卻只有較慢的吞吐速度。所以系統(tǒng)所使用的channel 組件是自定義的，以同時(shí)解決數(shù)據(jù)丟失和吞吐速度的問(wèn)題。

3.2 Kafka

Kafka 主要用于管理用戶(hù)訪(fǎng)問(wèn)日志數(shù)據(jù)（PV、UV 等），這些數(shù)據(jù)一般以日志方式存儲(chǔ)。將消息隊(duì)列技術(shù)應(yīng)用至數(shù)據(jù)處理，有助于提高實(shí)時(shí)日志系統(tǒng)的日志解析效率。而Kafka也包含三個(gè)組成部分：Producer、Topic、Consumer，如圖5所示：

圖5： Kafka 結(jié)構(gòu)示意

Producer：消息和數(shù)據(jù)的生產(chǎn)者，向Kafka 的一個(gè)topic發(fā)送消息。

Topic：如同它的名字一樣，topic 是一個(gè)將producer 發(fā)來(lái)的消息加以隔離分類(lèi)的部分，不同的消息將會(huì)被推入不同的topic 中，當(dāng)producer 向topic 發(fā)送消息時(shí)，就會(huì)向指定的topic 中傳入消息。

Costumer：消息和數(shù)據(jù)的消費(fèi)者，接收topic 的消息并處理其傳入的消息。

在 本 文 的flume 與Kafka 的 應(yīng) 用 中，producer 就 是flume。Flume 的最后一級(jí)sink 采集的數(shù)據(jù)流將數(shù)據(jù)發(fā)送給Kafka 的topic。使用檢測(cè)文件夾變化的Spool Source、高吞吐的MemoryChannel、自定義sink。

4 系統(tǒng)研究

4.1 系統(tǒng)簡(jiǎn)單設(shè)定

在～/flume/conf 文件夾下新建spool.conf，在spool.conf中進(jìn)行Spool Source 和自定義sink 的編寫(xiě)并將channel 設(shè)置為MemoryChannel：

4.2 自定義channel組件

雖然系統(tǒng)的整體已經(jīng)基本完成，但是我們?nèi)匀辉谑褂肕emoryChannel，之前提到過(guò)MemoryChannel 可以實(shí)現(xiàn)高速傳輸?shù)溆邢薜娜萘吭跀?shù)據(jù)傳輸高峰時(shí)刻或agent 中止時(shí)會(huì)造成數(shù)據(jù)丟失；FileChannel 有較大的容量在agent 中止后可以恢復(fù)數(shù)據(jù)，但卻只有較慢的吞吐速度。兩者并不能完全符合一個(gè)擁有高速數(shù)據(jù)流量的實(shí)時(shí)數(shù)據(jù)收集系統(tǒng)的需求，所以我們尋求將MemoryChannel 和FileChannel 各自的優(yōu)勢(shì)結(jié)合起來(lái)，為系統(tǒng)自定義一個(gè)channel。

4.3 另一種解決方式

在我瀏覽flume 的系統(tǒng)文件時(shí)，在flume-ng-channel 文件夾下，發(fā)現(xiàn)了一個(gè)flume-kafka-channel 的文件夾，這就是flume 系統(tǒng)提供的KafkaChannel，同時(shí)也成為該系統(tǒng)channel的另一個(gè)解決方案：利用flume 自帶的KafkaChannel 對(duì)MemoryChannel 進(jìn)行替換。而KafkaChannel 的優(yōu)勢(shì)在于source 讀取日志文件后，傳輸給KafkaChannel 后，Kafka Channel 直接將數(shù)據(jù)傳送給Kafka，不再需要sink，即使數(shù)據(jù)量很大時(shí)也可以正常工作，并且可以采用TailFile Source讀取日志文件。

5 實(shí)驗(yàn)結(jié)果

利用模擬生成日志如圖6 所示。

圖6： 用戶(hù)的訪(fǎng)問(wèn)日志

統(tǒng)計(jì)過(guò)去時(shí)間段的訪(fǎng)客和來(lái)源，如圖7、圖8 所示。

圖7： 對(duì)應(yīng)IP 以及訪(fǎng)客數(shù)量（次數(shù)）

圖8： 對(duì)應(yīng)網(wǎng)站的訪(fǎng)客數(shù)量（次數(shù)）

6 結(jié)論

Flume 所提供的多種source、channel 以及sink 組件，讓我們針對(duì)數(shù)據(jù)采集處理的各種不同情況有了更多的組合方式，同時(shí)也擁有更高的采集處理效率。Kafka 將數(shù)據(jù)安全地存儲(chǔ)在一個(gè)分布式、高可靠、高可用、高容錯(cuò)的集群中，通過(guò)連接、聚合、篩選、轉(zhuǎn)換等方式處理數(shù)據(jù)流。而Kafka 也可以作為flume 的channel 組件，由flume 的sink 將數(shù)據(jù)傳入HDFS、HBase、Hive 等數(shù)據(jù)庫(kù)或文件系統(tǒng)，在實(shí)現(xiàn)負(fù)載均衡的同時(shí)也實(shí)現(xiàn)數(shù)據(jù)在前期采集、預(yù)處理的加工。