Flume

1. 前言

　　flume是由cloudera軟件公司產出的可分布式日誌收集系統，後與2009年被捐贈了apache軟件基金會，為hadoop相關組件之一。尤其近幾年隨著flume的不斷被完善以及升級版本的逐一推出，特別是flume-ng;同時flume內部的各種組件不斷豐富，用戶在開發的過程中使用的便利性得到很大的改善，現已成為apache top項目之一.

2. 概述

2.1. 什麽是flume?
http://flume.apache.org/index.html

Apache Flume 是一個從可以收集例如日誌，事件等數據資源，並將這些數量龐大的數據從各項數據資源中集中起來存儲的工具

等集中存儲器中。其結構如下圖所示：

2.2. Flume特性

• Flume是一個分布式、可靠、和高可用的海量日誌采集、聚合和傳輸的系統。
• Flume可以采集文件，socket數據包、文件、文件夾、kafka等各種形式源數據，又可以將采集到的數據(下沈sink)輸出到HDFS、hbase、hive、kafka等眾多外部存儲系統中
• 一般的采集需求，通過對flume的簡單配置即可實現
• Flume針對特殊場景也具備良好的自定義擴展能力，因此，flume
  可以適用於大部分的日常數據采集場景

3. Flume原理

3.1. Flume組件詳解

對於每一個Agent來說,它就是一共獨立的守護進程(JVM),它從客戶端接收數據，如下圖所示flume的基本模型

1、 Flume分布式系統中最核心的角色是agent，flume采集系統就是由一個個agent所連接起來形成

2、 每一個agent相當於一個數據(被封裝成Event對象)傳遞員，內部有三個組件：

a) Source：采集組件，用於跟數據源對接，以獲取數據

b) Sink：下沈組件，用於往下一級agent傳遞數據或者往最終存儲系統傳遞數據

c) Channel：傳輸通道組件，用於從

sink

首先來看一下flume官網中對Event的定義

　　一行文本內容會被反序列化成一個event(序列化是將對象狀態轉換為可保持或傳輸的格式的過程。與序列化相對的是反序列化，它將流轉換為對象。這兩個過程結合起來，可以輕松地存儲和傳輸數據)，event的最大定義為2048字節，超過，則會切割，剩下的會被放到下一個event中，默認編碼是UTF-8。

3.2. Flume采集結構圖

3.2.1. 簡單結構

單個agent采集數據

3.2.2. 復雜結構

多級agent之間串聯

4. Flume實戰案例

4.1. Flume的安裝部署

1、Flume的安裝非常簡單，只需要解壓即可，當然，前提是已有hadoop環境

上傳安裝包到數據源所在節點上

然後解壓 tar -zxvf apache-flume-1.6.0-bin.tar.gz

然後進入flume的目錄，修改conf下的flume-env.sh，在裏面配置JAVA_HOME

2、根據數據采集的需求配置采集方案，描述在配置文件中(文件名可任意自定義)

3、指定采集方案配置文件，在相應的節點上啟動flume agent

先用一個最簡單的例子來測試一下程序環境是否正常

1、先在flume的conf目錄下新建一個配置文件（采集方案）

vi netcat-logger.properties

# 定義這個agent中各組件的名字
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 描述和配置source組件：r1
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# 描述和配置sink組件：k1
a1.sinks.k1.type = logger

# 描述和配置channel組件，此處使用是內存緩存的方式
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 描述和配置source  channel   sink之間的連接關系
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2、啟動agent去采集數據

-c conf 指定flume自身的配置文件所在目錄

-f conf/netcat-logger.conf 指定我們所描述的采集方案

-n a1 指定我們這個agent的名字

3、測試

先要往agent的source所監聽的端口上發送數據，讓agent有數據可采

隨便在一個能跟agent節點聯網的機器上

telnet anget-hostname port （telnet localhost 44444）

4.2. 采集案例

4.2.1. 采集目錄到HDFS

結構示意圖：

采集需求：某服務器的某特定目錄下，會不斷產生新的文件，每當有新文件出現，就需要把文件采集到HDFS中去

根據需求，首先定義以下3大要素

l 數據源組件，即source ——監控文件目錄 : spooldir

spooldir特性：

1、監視一個目錄，只要目錄中出現新文件，就會采集文件中的內容

2、采集完成的文件，會被agent自動添加一個後綴：COMPLETED

3、所監視的目錄中不允許重復出現相同文件名的文件

l 下沈組件，即sink——HDFS文件系統 : hdfs sink

l 通道組件，即channel——可用file channel 也可以用內存channel

配置文件編寫：

#定義三大組件的名稱
agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# 配置source組件
agent1.sources.source1.type = spooldir
agent1.sources.source1.spoolDir = /home/hadoop/logs/
agent1.sources.source1.fileHeader = false

#配置攔截器
agent1.sources.source1.interceptors = i1
agent1.sources.source1.interceptors.i1.type = host
agent1.sources.source1.interceptors.i1.hostHeader = hostname

# 配置sink組件
agent1.sinks.sink1.type = hdfs
agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M
agent1.sinks.sink1.hdfs.filePrefix = access_log
agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text
agent1.sinks.sink1.hdfs.rollSize = 102400
agent1.sinks.sink1.hdfs.rollCount = 1000000
agent1.sinks.sink1.hdfs.rollInterval = 60
#agent1.sinks.sink1.hdfs.round = true
#agent1.sinks.sink1.hdfs.roundValue = 10
#agent1.sinks.sink1.hdfs.roundUnit = minute
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in memory
agent1.channels.channel1.type = memory
agent1.channels.channel1.keep-alive = 120
agent1.channels.channel1.capacity = 500000
agent1.channels.channel1.transactionCapacity = 600

# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

Channel參數解釋：

capacity：默認該通道中最大的可以存儲的event數量

trasactionCapacity：每次最大可以從source中拿到或者送到sink中的event數量

keep-alive：event添加到通道中或者移出的允許時間

4.2.2. 采集文件到HDFS

采集需求：比如業務系統使用log4j生成的日誌，日誌內容不斷增加，需要把追加到日誌文件中的數據實時采集到hdfs

根據需求，首先定義以下3大要素

• 采集源，即source——監控文件內容更新 : exec ‘tail -F file’
• 下沈目標，即sink——HDFS文件系統 : hdfs sink
• Source和sink之間的傳遞通道——channel，可用file channel 也可以用 內存channel

配置文件編寫：

agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# Describe/configure tail -F source1
agent1.sources.source1.type = exec
agent1.sources.source1.command = tail -F /home/hadoop/logs/access_log
agent1.sources.source1.channels = channel1

#configure host for source
agent1.sources.source1.interceptors = i1
agent1.sources.source1.interceptors.i1.type = host
agent1.sources.source1.interceptors.i1.hostHeader = hostname

# Describe sink1
agent1.sinks.sink1.type = hdfs
#a1.sinks.k1.channel = c1
agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M
agent1.sinks.sink1.hdfs.filePrefix = access_log
agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text
agent1.sinks.sink1.hdfs.rollSize = 102400
agent1.sinks.sink1.hdfs.rollCount = 1000000
agent1.sinks.sink1.hdfs.rollInterval = 60
agent1.sinks.sink1.hdfs.round = true
agent1.sinks.sink1.hdfs.roundValue = 10
agent1.sinks.sink1.hdfs.roundUnit = minute
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true

# Use a channel which buffers events in memory
agent1.channels.channel1.type = memory
agent1.channels.channel1.keep-alive = 120
agent1.channels.channel1.capacity = 500000
agent1.channels.channel1.transactionCapacity = 600

# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

3、兩個agent級聯

4.3. 更多source和sink組件

Flume支持眾多的source和sink類型，詳細手冊可參考官方文檔

http://flume.apache.org/FlumeUserGuide.html

4.4. HA Flume配置案例

在完成單點的Flume NG搭建後，下面我們搭建一個高可用的Flume NG集群，架構圖如下所示：

　　圖中，我們可以看出，Flume的存儲可以支持多種，這裏只列舉了HDFS和Kafka（如：存儲最新的一周日誌，並給Spark Streaming系統提供實時日誌流。

4.4.1. 角色分配

Flume的Agent和Collector分布如下表所示：

圖中所示，Agent1，Agent2，Agent3數據分別流入到Collector1和Collector2，Flume NG本身提供了Failover機制，可以自動切換和恢復。在上圖中，有3個產生日誌服務器分布在不同的機房，要把所有的日誌都收集到一個集群中存儲。下面我們開發配置Flume NG集群。

4.4.2. 配置

在下面單點Flume中，基本配置都完成了，我們只需要新添加兩個配置文件，它們是agent.properties和collector.properties，其配置內容如下所示：

1、agent配置

vi conf/agent.properties

#agent1 name
agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2

#set gruop
agent1.sinkgroups = g1

#set channel
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100

agent1.sources.r1.channels = c1
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /root/log/test.log

agent1.sources.r1.interceptors = i1 i2
agent1.sources.r1.interceptors.i1.type = static
agent1.sources.r1.interceptors.i1.key = Type
agent1.sources.r1.interceptors.i1.value = LOGIN
agent1.sources.r1.interceptors.i2.type = timestamp

# set sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = mini2
agent1.sinks.k1.port = 52020

# set sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = mini3
agent1.sinks.k2.port = 52020

#set sink group
agent1.sinkgroups.g1.sinks = k1 k2

#set failover
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000

啟動命令：

bin/flume-ng agent -n agent1 -c conf -f conf/agent.properties -Dflume.root.logger=DEBUG,console

2、collector配置

vi collector.properties

#set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1

#set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = mini2
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = mini2
a1.sources.r1.channels = c1

#set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/home/hdfs/flume/logdfs
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

在mini3上，需要修改上述配置中的紅色字體主機名為mini3

啟動命令：

bin/flume-ng agent -n a1 -c conf -f conf/collector.properties -Dflume.root.logger=DEBUG,console

4.4.3. FAILOVER測試

下面我們來測試下Flume NG集群的高可用（故障轉移）。場景如下：我們在Agent1節點上傳文件，由於我們配置Collector1的權重比Collector2大，所以 Collector1優先采集並上傳到存儲系統。然後我們kill掉Collector1，此時有Collector2負責日誌的采集上傳工作，之後，我 們手動恢復Collector1節點的Flume服務，再次在Agent1上次文件，發現Collector1恢復優先級別的采集工作。具體截圖如下所 示：

Collector1優先上傳

HDFS集群中上傳的log內容預覽

Collector1宕機，Collector2獲取優先上傳權限

重啟Collector1服務，Collector1重新獲得優先上傳的權限

Flume可分布式日誌收集系統