flume--爲搬磚而生，日誌傳輸的一把好手

時間 2020-06-04

原文原文鏈接

（一）flume的產生

爲何會有flumejava

隨着互聯網的發展，人們對網絡日誌產生的信息也愈來愈重視。不只如此，咱們的服務器，好比Nginx，天天都會產生大量的日誌。咱們要將這些日誌收集到指定的地方，好比hdfs平臺，進行分析。可是大量的日誌產生的位置比較分散，可能來自於Tomcat、Nginx、甚至是數據庫等等，並且存儲的目的地也不同，這就致使了數據採集的複雜性。然鵝最關鍵的問題是，若是在採集的過程當中，出現問題了該怎麼辦？好比咱們要把日誌收集到hdfs平臺上進行分析，咱們能夠經過hdfs shell的方式，可是咱們沒法對日誌採集的情況進行監控。針對以上種種問題，flume就誕生了，所以看到這裏小夥伴也能猜到flume是幹什麼的了。python

flume是什麼golang

關於flume是什麼？官網給出了定義，咱們來看一下。由於flume是Apache的一個頂級項目，因此官網就是flume.apache.orgweb

flume是一個分佈式、高可靠、高可用的服務，用於高效地收集、聚合、移動大量的數據集。flume具備一個簡單、靈活並基於數據流的架構，它具有可調節的可靠機制和許多故障轉移以及恢復機制，並具備很強的健壯性和容錯性。經過使用一種簡單可擴展的數據模型來提供應用程序的在線分析。shell

咱們從圖中，能夠看到一個長方形，裏面圍住的部分就是flume的一個agent。咱們經過flume移動數據就至關於在寫一個配置文件。source：源。表示咱們要從哪裏收集，好比這裏來自於web服務器。sink：下沉。表示咱們要把數據沉到什麼地方去，這裏是hdfs。那麼中間還有一個channel是幹什麼的呢？channel：管道，咱們不會把剛收集到的日誌馬上sink到hdfs上面，那樣效率太慢，而是會先存到channel裏面，這個channel就是消息隊列，通常是kafka，當管道滿了，而後再把管道里面的數據一次性sink的目的地，這裏是hdfs。數據庫

所以通俗一點的說：apache

flume就是一個日誌採集工具，或者理解爲一個搬運工，將日誌從一個地方搬到另外一個地方去。特徵是：具有高可用、高可靠、分佈式處理模式。裏面定製了不少的source和sink，支持咱們從不一樣的源端讀取日誌文件，而後sink到不一樣的地方去。數組

flume的特徵緩存

flume能夠高效率地從多個服務器中收集日誌並存儲到hdfs/hbase中安全
除了日誌信息，flume也能夠用來接入收集規模宏大的社交網絡節點事件數據，好比Facebook、Twitter等等
支持各類不一樣數據源，並可以輸送到不一樣的地方

支持多路徑，多管道輸入，多管道輸出，上下文路由等等

什麼意思呢？意思就是咱們sink的時候，也能夠做爲下一個agent的source，不必定要是hdfs等平臺，而且多個sink還能夠到同一個source裏面。

同理，一個source也能夠到多個channel，而後sink到多個地方

能夠水平擴展

flume的應用場景

flume其實還能夠有其餘的功能，好比數據的處理，可是在大數據實際應用中，flume基本上只用於日誌收集工做，根據不一樣的業務場景，編寫不一樣的配置文件。

（二）flume的client

flume的客戶端

什麼是flume的客戶端呢？實際上，部署flume的機器就能夠稱之爲一個客戶端。部署flume，是否是須要一些環境呢？

根據官網的要求，至少要知足如下幾點

1.jdk1.8或更高，大數據的這些組件都和java脫不開關係
2.內存。由於flume是基於流式的，所以內存也是必須的
3.磁盤空間，咱們不會把數據一直保存到內存裏面，最終要刷到磁盤上，所以磁盤空間也是必須的。
4.目錄權限，你要往一個目錄裏面寫文件的話，總要有寫或者讀的權限吧

感受二、三、4都是廢話，這不是顯然的嗎？

（三）flume的source

咱們以前說過，一個agent就至關於一個配置文件，裏面有三劍客，分別是source、channel、sink。經過編寫配置文件，配置source、channel、sink，來將數據從一端傳輸到另外一端。
咱們來看看source。

source就是咱們的源端，從不一樣的數據源讀取日誌信息放到channel裏，而後sink取出channel裏面的日誌信息到其餘的地方，好比hdfs、hbase等等。

所以flume的架構很好理解，就這三個組成部分。既然flume的source支持不一樣的數據源，那麼咱們就來看看，它都支持哪些不一樣的數據源呢？

source支持的數據源

支持如下幾種source

Avro Source

flume第一種數據來源，能夠監聽IP和端口，並不必定是本機的IP，能夠監聽其餘機器的IP和port，用於獲取數據。從外部的Avro client streams接收events，當與另外一個flume agent內置的avro sink配對時，能夠建立分層蒐集的拓撲結構。感受官網說的好陽春白雪，其實說白了，仍是下面這張圖，一個agent的source能夠是上一個agent的sink，之間經過avro去鏈接

字體加粗的屬性必須進行設置

由於source的類型是avro，所以type也要指定爲avro
Thrift Source
```
和avro source基本相似，可是avro能夠支持ip過濾，可是thrift不支持。
```
由於source是thrift，因此type要改爲thrift。是什麼樣的source，那麼type就改爲什麼。
Exec Source

比較經常使用的一種source，能夠運行Unix命令，而後該命令會一直執行，讀取文件，做爲flume數據的來源，須要配置相應的channel、type、以及運行的命令。而且根據官網介紹，本方式不保證數據百分百會傳送到指定的位置，若是出現問題，數據可能丟失。可使用Spooling Directory等其餘手段，保證數據傳輸
JMS Source

本方式會從JMS消息隊列的目的地，好比一個隊列或者一個topic中讀取數據；須要在plugins.d即插件目錄下放置相應的jar包；不過此來源通常用的較少，由於侷限性比較強。
Spooling Directory Source

Spooling Directory Source監測配置的目錄下新增的文件，並將文件中的數據讀取出來。其中，Spool Source有2個注意地方，第一個是拷貝到spoolDir指定的目錄下的文件不能夠再打開編輯，第二個是spool目錄下不可包含相應的子目錄。這個主要用途做爲對日誌的準實時監控。

加粗的表示必須填寫，其中spooldir，就是咱們指定的目錄，固然type也要指定爲spooldir

自定製source

其實flume提供的source仍是不能徹底知足咱們的需求，所以咱們也能夠自定義source。

固然還有其它source，不過不經常使用。這裏可能有點抽象，先來清楚一下概念，瞭解一下flume的架構、各個組成部分的功能，後續會進行演示

（四）flume的channel

仍是這張放了好幾遍的圖，source咱們已經介紹過了，下面介紹channel。
channel咱們其實咱們以前也說過了，是一個管道，相似於golang裏面的channel。或者咱們把它理解爲一個緩存也行，source收集過來的日誌並不直接sink到hdfs上面，而是會放到緩存裏面，等緩存滿了再一次性sink到指定的地方。
爲何要這麼設計呢？介紹source收集日誌的速度很快，可是sink的速度很慢，若是source直接對接sink的話，那麼就會致使讀寫速度不一致了，這就跟雙十一要引入消息隊列同樣，若是上游的訂單系統直接對接下游的庫存系統，那麼上游爆發的流量很容易將下游沖垮，因此才須要引入消息隊列進行削峯填谷，二者的道理是相似的。

關於緩存，有兩種。一種是基於內存的緩存，另外一種是基於磁盤的緩存。
基於內存的緩存：速度快，可是不安全，並且容量有限
基於磁盤的緩存：可靠性高，容量大，可是讀寫速度會慢一些。

flume官方爲咱們提供了以下channel

基於內存的channel，咱們不多使用，由於數據容易丟失。File channle和kafka channel很經常使用

好比file channel，那麼這裏的type要指定爲file，由於這裏的緩存使用文件來實現的。配置的話能夠指定大小

（五）flume的sink

咱們以前說過，source是用於收集數據，channel是用於緩存數據，sink是用於將數據寫到外部的存儲設備。那咱們外部的存儲設備不同，那麼sink的類型就會不同，所以sink實際上也是可定製的，可配置的。對於不一樣的外部地址，那麼sink就有不一樣的實現。

flume也爲咱們提供了不少sink，好比hdfs sink、hive sink等等。固然還有avro sink，這個在以前介紹過，是爲了對接下一個agent的avro source

固然每個sink都有不少額外的配置，好比文件的前綴啊、回滾間隔、回滾策略啊等等

（六）flume的interceptor

flume的攔截器，這是什麼？

flume能夠在移動數據時修改/刪除事件，這是在攔截器的幫助下完成的。全部的攔截器都是實現了org.apache.flume.interceptor.Interceptor接口的類，攔截器能夠基於開發人員所選擇的任意條件來修改、甚至刪除事件。而且flume還支持鏈式的攔截器(補充：意思就是能夠設置多個攔截器，而後順序的調用。)，這能夠經過在配置中指定多個攔截器的類名組成的列表來實現。攔截器在源配置中經過一系列的空格進行分隔，攔截器被指定的順序也是它們被調用的順序，每個攔截器所返回的事件在列表中會鏈式的傳遞到下一個攔截器，而且能夠修改或者刪除事件。若是一個攔截器須要刪除事件，那麼它只須要在列表中將該返回的事件不進行返回便可。若是要刪除全部事件，只須要簡單地返回一個空列表便可。攔截器是一個命名組件，如下是如何經過配置建立攔截器的一個例子。

攔截器主要用於source端，咱們能夠將獲取的數據攔截下來，而後加上一個鍵值對，能夠指定文件的名字、來源、時間戳等等，而後sink到不一樣的地方去。好比咱們收集了不少日誌，好比登錄日誌、行爲日誌等等，那麼咱們在攔截的時候，即可以指定哪些日誌都是什麼類型，而後輸出的時候，不一樣的日誌輸出到不一樣的地方去

（七）flume的selector

咱們的source在收集完數據以後要寫到哪個channel裏面呢？好比咱們蒐集的日誌有三種，一種是隻包含字母，另外一種只包含數字，最後一種只包含符號(假設是這樣蛤)。hdfs上面有三個文件，正好對應三種日誌類型。如今咱們的問題是，不一樣類型的日誌怎麼存到不一樣的文件裏面去呢？既然要存到不一樣的文件裏面，那麼首先要存到不一樣的channel裏面，channel1只包含字母、channel2只包含數字、channel3只包含符號，而後sink1對接channel一、sink2對接channel二、sink3對接channel3，理論上狀況是這樣的。

可是如今問題又來了，怎麼存到不一樣的channel裏面呢？這個時候選擇器就出現了

官網提供了哪些selector呢

Replicating Channel Selector (default)

這也是默認的channel，type指定爲replicating。意思是數據會複製到每個channel中。

Multiplexing Channel Selector

復分模式的channel，不會把數據拷貝了，而是將數據總體分發到不一樣的channel中，固然要指定type爲multiplexing。能夠看到無論是source、channel、sink仍是這裏的selector，都有不一樣的類型，而後在各自的type中指定便可。

Custom Channel Selector

本身實現的channel，不一樣的數據分發到不一樣的channel中，至關於對Multiplexing Channel Selector進行了自定製，這樣就能夠按照咱們本身的邏輯來決定數據分發到哪個channel當中。

（八）flume的event

例如一些消息隊列，發送的數據並非原生的數據，而是對數據要進行一個封裝，咱們能夠經過python往activemq裏面發送數據，而後使用golang去接。可是python最終發到mq裏面不是一個簡單的數據，golang接收的也不是一個簡單的數據，而是對數據進行了一個封裝。同理flume也是同樣，在flume中，有一個event，咱們好像在攔截器的時候提到過，那麼它是什麼呢？event在flume當中叫作事件,flume在傳輸日誌的時候，每一條數據都會被封裝成一個event。

爲何要有event呢？

首先咱們能夠想一下，數據格式有不少種，source也不一樣，channel不一樣、sink也各不相同，那麼我如何才能將其統一塊兒來呢？這個時候就須要event了，將每一條記錄都封裝爲你們都認識event不就好了嗎？這就是爲何要有event。固然event主要包含兩個部分，一個是header，一個body。body是一個字節數組，裏面記錄了日誌信息，header則是一個hashmap，咱們能夠添加一些本身的信息，若是隻有body沒有header的話，那麼咱們的信息就會打到body裏面，這樣就對咱們要傳送的數據形成污染了，具備侵入性，因此要有header

（九）flume的agent

agent是flume的運行核心，是flume運行的最小單位。一個flume的agent就是包含了source、channel、sink的一個jvm進程，用於日誌的收集和傳輸。

首先咱們到目前爲止介紹的都是概念，其實flume的概念也很好理解，主要就是各個組件的那些配置尚未介紹，後面會介紹，目前先了解flume的基礎架構便可。所以以前也說過，用flume傳輸日誌，就至關於在指定agent中各個組件的配置。一個agent能夠包含多個source、多個channel、多個sink

A Flume event is defined as a unit of data flow having a byte payload and an optional set of string attributes. A Flume agent is a (JVM) process that hosts the components through which events flow from an external source to the next destination (hop).
一個flume事件被定義爲一個數據流單元，具備字節負載和可選的字符屬性集。一個flume agent就是一個jvm進程，承載着各個組件(source、channel、sink)，事件能夠經過這些組件從外部源流向下一個目的地。

固然不要忘記，agent仍是能夠串聯的，一個agent的sink對接下一個agent的source

（十）flume的安裝

flume下載地址：http://archive.cloudera.com/cdh5/cdh/5
選擇flume便可，爲何不到官網下載，由於cdh版本的話能夠避免不少問題，好比jar包的衝突。
所以大數據組件安裝的話，推薦去這個網站下載，不建議使用社區版本的。

我用的是阿里雲的雲服務器，全部大數據相關的軟件都放在/opt目錄下，所以咱們在/opt下面新建一個flume目錄，而後把相應的安裝包放進去解壓

解壓完成以後，咱們來看看flume的結構

咱們進入conf目錄下，修改一下配置。若是咱們只修改一個配置的話，那麼須要修改什麼呢？不用想，確定是jdk的路徑。對，咱們須要在flume-env.sh中指定JAVA_HOME,可是進入目錄發現沒有這個文件，可是有一個flume-env.sh.template，這是模板，咱們用這個模板生成一個便可

而後將flume加到環境變量裏面去

export FLUME_HOME=/opt/flume/apache-flume-1.6.0-cdh5.7.5-bin
export PATH=$FLUME_HOME/bin:$PATH

而後source一下

輸入flume-ng version查看版本，和其餘框架不同，flume的話須要加上-ng，至於緣由後面會說。

（十一）flume的兩種版本

flume分爲兩個版本，一個1.x以前的版本，叫作flume og，另外一個是1.x版本，叫作flume ng
至於差異就不介紹了，總之如今用的話就是用flume 1.x版本的就好了。

（十二）flume的avro_source

任務：使用avro_source收集數據，打印到控制檯

終於到實戰部分了，咱們下面就經過編寫配置文件的方式來移動數據，以前也說了，使用flume就是編寫配置文件。
至於channel，咱們就無所謂了，使用內存channel仍是磁盤channel都無所謂，咱們這裏主要是介紹source。而sink則就是控制檯

那麼配置怎麼寫呢？

# 這裏的a1就是咱們agent的名字
a1.sources = r1  #  命名agent的source爲r1
a1.channels = c1  # 同理c1是channel的名字
a1.sinks = k1  # 同理k1是sink的名字

a1.sources.r1.type = avro  # 配置r1的的類型avro
a1.sources.r1.bind = localhost  # 綁定的端口爲本地
a1.sources.r1.port = 4141  # 端口爲4141

a1.channels.c1.type = memory  # 指定c1的類型爲內存
a1.channels.c1.capacity = 1000  # 指定c1的容量爲1000個event
a1.channels.c1.transcactionCapacity = 1000  # 指定c1的每次處理容量爲1000個event

a1.sinks.k1.type = logger  # 表示不產生實體文件，就打印在控制檯

# 將source和sink與channel綁定起來
a1.sources.r1.channels = c1  # source r1的channel指定爲c1 
a1.sinks.k1.channel = c1  # sink k1的channel指定爲c1。可是這裏再也不是channels，而是channel，要注意

咱們隨便在一個目錄，新建一個文件，將上面的內容去掉註釋而後寫進去

那麼如何啓動flume呢？
flume-ng agent -c $FLUME_HOME/conf -f 咱們寫的配置文件的路徑 -n a1 -Dflume.root.logger=INFO,console

參數解釋：
-c $FLUME_HOME/conf：首先flume啓動是有默認配置的，這個配置就在$FLUME_HOME/conf下面
-f 咱們寫的配置文件的路徑：這個很好理解，咱們本身寫的配置能夠經過-f來指定
-n a1：咱們在配置文件中指定的agent名字
-Dflume.root.logger=INFO,console：表示flume自身運行狀態的日誌，按需配置，詳細信息控制檯打印

咱們啓動服務，因爲咱們本身的conf就在當前目錄下，因此直接flume-ng agent -c $FLUME_HOME/conf -f ./conf -n a1 -Dflume.root.logger=INFO,console啓動便可

服務已經啓動了,監聽本地的4141端口，那麼咱們就往這個端口裏面發送數據。

發送數據，咱們能夠發送一個文件，可使用flume自帶的一個方法。
flume-ng avro-client -c $FLUME_HOME/conf -H localhost -p 4141 -F ~/.bashrc

avro-client：建立一個客戶端
-c $FLUME_HOME/conf：指定配置，剛纔說了
-H localhost：指定目的ip
-p 4141：指定目的端口
-F ~/.bashrc：指定發送的文件

顯示內容已經接受到了，會將文件打開，每一條都是一個event，包含header(咱們這裏沒有指定)和event(是一個字節數組)

（十三）flume的exec_source

下面來看看flume的exec_source，flume之因此有這麼多source、channel、sink，主要是爲了解決不一樣場景的業務，那麼這個exec_source是用來幹什麼的呢？

exec_source能夠實時監控一個文本文件，若是文件有變化，能夠打印出來

下面就來寫配置文件，在當前目錄新建一個conf1，寫上以下內容(不含註釋)

a1.sources = r1  #  命名agent的source爲r1
a1.channels = c1  # 同理c1是channel的名字
a1.sinks = k1  # 同理k1是sink的名字

a1.sources.r1.type = exec  # 配置r1的的類型exec
# a1.sources.r1.bind = localhost  # 綁定的端口爲本地
# a1.sources.r1.port = 4141  # 端口爲4141
# 因爲是監控一個文件，因此不須要在綁定ip和端口了
a1.sources.r1.command = tail -F /root/mashiro.txt  # 指定命令，監控這個文件

a1.channels.c1.type = memory  # 指定c1的類型爲內存
a1.channels.c1.capacity = 1000  # 指定c1的容量爲1000個event
a1.channels.c1.transcactionCapacity = 1000  # 指定c1的每次處理容量爲1000個event

a1.sinks.k1.type = logger  # 表示不產生實體文件，就打印在控制檯

# 將source和sink與channel綁定起來
a1.sources.r1.channels = c1  # source r1的channel指定爲c1 
a1.sinks.k1.channel = c1  # sink k1的channel指定爲c1。可是這裏再也不是channels，而是channel，要注意

啓動flume，flume-ng agent -c $FLUME_HOME/conf -f ./conf1 -n a1 -Dflume.root.logger=INFO,console

此時服務啓動成功，可是尚未東西過來

（十四）file channel

基於內存的channel，咱們在介紹source的時候，演示過了，下面來看看基於文件的channel。

a1.sources = r1  
a1.channels = c1 
a1.sinks = k1  

a1.sources.r1.type = exec 
a1.sources.r1.command = tail -F /root/mashiro.txt  

a1.channels.c1.type = file  # 指定類型爲file
a1.channels.c1.checkpointDir = /root/checkPoint  # checkpoint將被存儲的目錄
a1.channels.c1.dataDirs = /root/data  # 數據的存儲位置

a1.sinks.k1.type = logger  # 表示不產生實體文件，就打印在控制檯

# 將source和sink與channel綁定起來
a1.sources.r1.channels = c1  # source r1的channel指定爲c1 
a1.sinks.k1.channel = c1  # sink k1的channel指定爲c1。可是這裏再也不是channels，而是channel，要注意

先來看看/root目錄的內容

而後執行命令,flume-ng agent -c $FLUME_HOME/conf -f ./conf2 -n a1 -Dflume.root.logger=INFO,console

再來查看當前目錄

能夠看到checkPoint和data兩個目錄，並且這是flume幫咱們建立的,咱們來往文件裏面寫點東西

（十五）hdfs sink

下面介紹sink，這裏介紹一下hdfs sink，由於收集過來的日誌主要放到hdfs平臺上面。
既然要把收集的數據放大hdfs上面，那麼flume就必須能鏈接hdfs，須要把依賴的jar包，放到$FLUME_HOME/lib下面，須要哪些jar包呢？又從哪裏找呢？咱們能夠從$HADOOP_HOME/share/hadoop目錄裏面找。
commons-configuration-x.x.jar  
hadoop-auth-x.x.x.jar
hadoop-common-x.x.x.jar
hadoop-hdfs-x.x.x.jar
# jar前面的x表示版本號，在找的時候可使用模糊查詢，好比我如今在$HADOOP_HOME/share/hadoop目錄下，要找commons-configuration-x.x.jar  的話，即可以經過find ./ -name commons-configuration*便可。

若是你的flume版本是1.99的話，那麼還須要如下兩個jar包，不然的話就不須要
commons-io-x.x.jar
htrace-core-.x.x.x-incubating.jar

找到以後，cp到$FLUME_HOME/lib下面

下面就是編寫配置文件

# 給source、channel、sink指定名字
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# source的配置
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/test.log

# channel的配置
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# sink的配置
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://localhost:9000
# 這條配置是給文件起一個前綴
a1.sinks.k1.hdfs.prefix = hive-  

# 綁定channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。