Flume的概述

Flume是Cloudera提供的一個高可用的，高可靠的，分布式的海量日 志采集、聚合和傳輸的系統。Flume基于流式架構，靈活簡單。

• 高可用(HA) flume框架（故障轉移機制）
• 高可靠 數據采集的可靠性
• 分布式 分布式集群搭建

Flume的作用

最主要的作用：實時讀取服務器本地磁盤的數據，將數據寫到HDFS、Kafka

Flume的優點

可以和任意存儲進程集成。

• 支持不同的采集源
• 支持多類型的目標源

輸入的的數據速率大于寫入目的存儲的速率，flume會進行緩沖，減小 hdfs的壓力。

flume中的事務基于channel，使用了兩個事務模型（sender + receiver），確保消息被可靠發送。

Flume使用兩個獨立的事務分別負責從soucrce到channel，以及從 channel到sink的事件傳遞。一旦事務中所有的數據全部成功提交到 channel，那么source才認為該數據讀取完成。同理，只有成功被sink 寫出去的數據，才會從channel中移除。

Flume的組成結構

1、Flume組成架構

2、Agent

a、簡介

Agent是一個JVM進程，它以事件的形式將數據從源頭送至目的。Agent 主要有3個部分組成，Source、Channel、Sink。

b、Source

Source是負責接收數據到Flume Agent的組件。Source組件可以處理 各種類型、各種格式的日志數據，包括avro、thrift、exec、jms、 spooling directory、netcat、sequence generator、syslog、 http、legacy。

c、Channel

Channel是位于Source和Sink之間的緩沖區。因此，Channel允許 Source和Sink運作在不同的速率上。Channel是線程安全的，可以同 時處理幾個Source的寫入操作和幾個Sink的讀取操作。 Flume自帶兩種Channel：Memory Channel和File Channel。 Memory Channel是內存中的隊列。Memory Channel在不需要關心 數據丟失的情景下適用。如果需要關心數據丟失，那么Memory Channel就不應該使用，因為程序死亡、機器宕機或者重啟都會導致數 據丟失。File Channel將所有事件寫到磁盤。因此在程序關閉或機器宕 機的情況下不會丟失數據。

d、Sink

Sink不斷地輪詢Channel中的事件且批量地移除它們，并將這些事件批 量寫入到存儲或索引系統、或者被發送到另一個Flume Agent。Sink 是完全事務性的。在從Channel批量刪除數據之前，每個Sink用 Channel啟動一個事務。批量事件一旦成功寫出到存儲系統或下一個 Flume Agent，Sink就利用Channel提交事務。事務一旦被提交，該 Channel從自己的內部緩沖區刪除事件。Sink組件目的地包括hdfs、 logger、avro、thrift、ipc、file、null、HBase、solr、自定義。

e、Event

傳輸單元，Flume數據傳輸的基本單元，以事件的形式將數據從源頭送 至目的地。 Event由可選的header和載有數據的一個byte array 構成。Header是容納了key-value字符串對的HashMap。

Flume agent的配置文件

單數據源單出口案例

這種模式是將多個flume給順序連接起來了，從最初的source開始到最 終sink傳送的目的存儲系統。此模式不建議橋接過多的flume數量， flume數量過多不僅會影響傳輸速率，而且一旦傳輸過程中某個節點 flume宕機，會影響整個傳輸系統。

flume實現監控端口數據案例：

用netcat工具向本機端口號：44444發送消息，flume監聽

# Name the components on this agent
# r1:表示a1的輸入源	a1：表示agent的名稱
a1.sources = r1
# k1：表示a1的輸出目的地
a1.sinks = k1
# c1：表示a1的緩沖區
a1.channels = c1

# Describe/configure the source
# 表示a1的輸入源類型為netcat端口類型
a1.sources.r1.type = netcat
# 表示a1的監聽主機
a1.soucres.r1.bind = localhost
# 表示a1的監聽的端口號
a1.sources.r1.port = 44444

# Describe the sink
# 表示a1的輸出目的地是控制臺logger類型
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
# 表示a1的channel類型是memory內存型
a1.channels.c1.type = memory
# 表示a1的channel總容量1000個event
a1.channels.c1.capacity = 1000
# 表示a1的channel傳輸時收集到100條event以后再去提交事務
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
# 表示講r1和c1鏈接起來
a1.sources.r1.channels = c1
# 表示將k1和c1鏈接起來
a1.sinks.k1.channel = c1

啟動flume：

• 方法一：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf - Dflume.root.logger=INFO,console

• 方法二：bin/flume-ng agent -c conf/ -n a1 –f job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console

參數說明：

• --conf conf/ ：表示(conf)配置文件存儲在conf/目錄
• --name a1 ：表示給agent起名為a1
• --conf-file job/flume-netcat.conf ：flume本次啟動讀取的配置 文件是在job文件夾下的flume-telnet.conf文件。
• -Dflume.root.logger==INFO,console ：-D表示flume運行時動 態修改flume.root.logger參數屬性值，并將控制臺日志打印級別設 置為INFO級別。日志級別包括:log、info、warn、error。

實時采集文件到HDFS上案例

用flume實時監聽某文件，當該文件的內容變化時，上傳該數據到HDFS上。

# Name the components on this agent
a2.sources = r2
a2.sinks = k2
a2.channels = c2

# Describe/configure the source
# 定義數據源文件的類型
a2.sources.r2.type = exec
# 監聽該目錄下的access.log文件
a2.sources.r2.command = tail -F /home/hadoop/nginx/logs/access.log
a2.sources.r2.shell = /bin/bash -c

# Describe the sink
a2.sinks.k2.type = hdfs
# 上傳文件的路徑 %Y%m%d為時間戳，自動生成對應時間 年月日
a2.sinks.k2.hdfs.path = hdfs://192.168.137.128:9000/flume/%Y%m%d/%H
#上傳文件的前綴
a2.sinks.k2.hdfs.filePrefix = logs-
#是否按照時間滾動文件夾
a2.sinks.k2.hdfs.round = true
#多少時間單位創建一個新的文件夾
a2.sinks.k2.hdfs.roundValue = 1
#重新定義時間單位
a2.sinks.k2.hdfs.roundUnit = hour
#是否使用本地時間戳
a2.sinks.k2.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a2.sinks.k2.hdfs.batchSize = 1000
#設置文件類型，可支持壓縮
a2.sinks.k2.hdfs.fileType = DataStream
#多久生成一個新的文件
a2.sinks.k2.hdfs.rollInterval = 60
#設置每個文件的滾動大小
a2.sinks.k2.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a2.sinks.k2.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2

實時讀取目錄文件到HDFS上案例

使用flume實時監聽整個目錄文件，當該目錄文件新增時，上傳該文件到HDFS上。

a3.sources = r3
a3.sinks = k3
a3.channels = c3

# Describe/configure the source
# 定義source類型為目錄
a3.sources.r3.type = spooldir
# 定義監控目錄
a3.sources.r3.spoolDir = /home/hadoop/bigdatasoftware/flume/upload
# 定義文件上傳完的后綴名
a3.sources.r3.fileSuffix = .COMPLETED
# 是否有五年間頭
a3.sources.r3.fileHeader = true
#忽略所有以.tmp結尾的文件，不上傳
a3.sources.r3.ignorePattern = ([^ ]*.tmp)

# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://192.168.137.128:9000/flume/upload/%Y%m%d/%H
#上傳文件的前綴
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照時間滾動文件夾
a3.sinks.k3.hdfs.round = true
#多少時間單位創建一個新的文件夾
a3.sinks.k3.hdfs.roundValue = 1
#重新定義時間單位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地時間戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a3.sinks.k3.hdfs.batchSize = 100
#設置文件類型，可支持壓縮
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一個新的文件
a3.sinks.k3.hdfs.rollInterval = 60
#設置每個文件的滾動大小大概是128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100

# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3

單數據源多出口案例（選擇器）

Flume支持將事件流向一個或者多個目的地。這種模式將數據源復制到 多個channel中，每個channel都有相同的數據，sink可以選擇傳送的 不同的目的地。

flume1監控文件的變動，并將變動的內容傳遞給flume2和flume3。

flume2負責輸出到HDFS上

flume3負責輸出到本地上

三個flume在同一臺設備上

flume1：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# 將數據流復制給所有channel
a1.sources.r1.selector.type = replicating

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/bigdatasoftware/nginx/logs/access.log
a1.sources.r1.shell = /bin/bash -c

# Describe the sink
# sink端的avro是一個數據發送者
a1.sinks.k1.type = avro
# 設置其中一個flume接收的地址
a1.sinks.k1.hostname = 192.168.137.128
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
# 設置另一個flume的接收地址
a1.sinks.k2.hostname = 192.168.137.128
a1.sinks.k2.port = 4142

# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

flume2:

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1

# Describe/configure the source
# source端的avro是一個數據接收服務
a2.sources.r1.type = avro
# 設置本機地址，注意端口號
a2.sources.r1.bind = 192.168.137.128
a2.sources.r1.port = 4141

# Describe the sink
a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path =
hdfs://192.168.137.128:9000/flume2/%Y%m%d/%H
#上傳文件的前綴
a2.sinks.k1.hdfs.filePrefix = flume2-
#是否按照時間滾動文件夾
a2.sinks.k1.hdfs.round = true
#多少時間單位創建一個新的文件夾
a2.sinks.k1.hdfs.roundValue = 1
#重新定義時間單位
a2.sinks.k1.hdfs.roundUnit = hour
#是否使用本地時間戳
a2.sinks.k1.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a2.sinks.k1.hdfs.batchSize = 100
#設置文件類型，可支持壓縮
a2.sinks.k1.hdfs.fileType = DataStream
#多久生成一個新的文件
a2.sinks.k1.hdfs.rollInterval = 600
#設置每個文件的滾動大小大概是128M
a2.sinks.k1.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a2.sinks.k1.hdfs.rollCount = 0

# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

flume3:

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2

# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = 192.168.137.128
a3.sources.r1.port = 4142

# Describe the sink
a3.sinks.k1.type = file_roll
a3.sinks.k1.sink.directory = /opt/module/data/flume3

# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100

# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2

注意：接收方與發送方的地址和端口號要對應

單數據源多出口案例（Sink組）

Flume支持使用將多個sink邏輯上分到一個sink組，flume將數據發送 到不同的sink，主要解決負載均衡和故障轉移問題。

配置1個接收日志文件的source和1個channel、兩個sink，分別輸送給flume-flume-console1和flume-flume-console2。

flume1：

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

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 22222

#定義一個sink組
#一個channel對應多個sink時要設置一個sinkgroups
a1.sinkgroups = g1
#指明sink組中的sink實例
a1.sinkgroups.g1.sinks = k1 k2
#設置sinkProcessor的類型（負載均衡）
a1.sinkgroups.g1.processor.type = load_balance
#①random-隨機分配  ②round_robin-輪循
a1.sinkgroups.g1.processor.selector = random


a1.channels.c1.type = memory


a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.137.128
a1.sinks.k1.port = 33333

a1.sinks.k2.type = avro
a1.sinks.k2.hostname = 192.168.137.129
a1.sinks.k2.port = 44444


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

flume2：

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

a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.137.128
a1.sources.r1.port = 33333

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000

a1.sinks.k1.type = logger

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

flume3：

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

a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.137.129
a1.sources.r1.port = 44444

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000

a1.sinks.k1.type = logger


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

多數據源匯總

這種模式是我們最常見的，也非常實用，日常web應用通常分布在上百 個服務器，大者甚至上千個、上萬個服務器。產生的日志，處理起來也 非常麻煩。用flume的這種組合方式能很好的解決這一問題，每臺服務 器部署一個flume采集日志，傳送到一個集中收集日志的flume，再由 此flume上傳到hdfs、hive、hbase、jms等，進行日志分析

flume1監控一個文件的變動

flume2監控一個端口的數據

flume1和flume2將數據發送給flume3，flume3最終將數據打印到控制臺。

flume1：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/nginx/logs/access.log
a1.sources.r1.shell = /bin/bash -c

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.137.129
a1.sinks.k1.port = 4141

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

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

flume2:

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1

# Describe/configure the source
a2.sources.r1.type = netcat
a2.sources.r1.bind = 198.168.137.128
a2.sources.r1.port = 44444

# Describe the sink
a2.sinks.k1.type = avro
a2.sinks.k1.hostname = 192.168.137.129
a2.sinks.k1.port = 4141

# Use a channel which buffers events in memory
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

flume3:

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c1

# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = 192.168.137.129
a3.sources.r1.port = 4141

# Describe the sink
# Describe the sink
a3.sinks.k1.type = logger

# Describe the channel
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a3.sources.r1.channels = c1
a3.sinks.k1.channel = c1