李柯

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

1 引言

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

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

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

在當(dāng)今的互聯(lián)網(wǎng)世界，日志文件的分析變得越來越重要。為了分析日志數(shù)據(jù)，我們利用網(wǎng)絡(luò)挖掘的幫助從這些日志數(shù)據(jù)中獲得所需的知識。而日志文件生成的速度極快，一天之內(nèi)，一個數(shù)據(jù)中心就能生成數(shù)十TB 的日志數(shù)據(jù)。傳統(tǒng)的數(shù)據(jù)采集數(shù)據(jù)來源于單一，且儲存、管理方式較為復(fù)雜，解析的數(shù)據(jù)量也相對較小，而現(xiàn)代大數(shù)據(jù)采集來源廣泛、數(shù)據(jù)量極大，數(shù)據(jù)類型繁雜，需要分布式的數(shù)據(jù)庫存儲。顯然傳統(tǒng)的數(shù)據(jù)采集方式無法適應(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í)也支持離線分析系統(tǒng)，并能在數(shù)據(jù)量增加的場合，通過增加節(jié)點(diǎn)數(shù)量水平拓展。

2 Flume簡介

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ā)送方、對數(shù)據(jù)進(jìn)行簡單處理交付數(shù)據(jù)接收分析方。

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

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

圖3： Flume 模型

3 Flume+Kafka

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

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

Kafka 作為中間件，一個突出的優(yōu)點(diǎn)就是將各層解耦，使得出錯時(shí)由于其封裝性不會影響其他部分工作，并不影響其他功能。數(shù)據(jù)從數(shù)據(jù)源中被捕獲，到flume 中采集分類，再到Kafka 中處理分類時(shí)，數(shù)據(jù)一方面能夠同步到HDFS 進(jìn)行離線運(yùn)算，另一方面也能夠進(jìn)行實(shí)時(shí)運(yùn)算，實(shí)現(xiàn)數(shù)據(jù)的并發(fā)操作。同時(shí)Kafka 的開箱即用式連接接口還整合了數(shù)百個事件源和事件接收器，使得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，共包含以下三個組件：source、channel、sink。但是組件中的channel 如使用MemoryChannel，雖然可以實(shí)現(xiàn)高速傳輸?shù)溆邢薜娜萘吭跀?shù)據(jù)傳輸高峰時(shí)刻或agent 中止時(shí)會造成數(shù)據(jù)丟失；如使用FileChannel，有較大的容量在agent 中止后可以恢復(fù)數(shù)據(jù)，但卻只有較慢的吞吐速度。所以系統(tǒng)所使用的channel 組件是自定義的，以同時(shí)解決數(shù)據(jù)丟失和吞吐速度的問題。

3.2 Kafka

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

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

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

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

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

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

4 系統(tǒng)研究

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

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

4.2 自定義channel組件

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

4.3 另一種解決方式

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

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

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

圖6： 用戶的訪問日志

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

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

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

6 結(jié)論

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