Flume 定義

　　Flume 是Cloudera 提供的一個高可用的，高可靠的，分散式的海量日誌採集、聚合和傳輸的系統。Flume 基於流式架構，靈活簡單。最主要的作用就是，實時讀取伺服器本地磁碟的資料，將資料寫入到HDFS

基礎框架

1 Agent

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

2 Source

Source 是負責接收資料到 Flume Agent 的元件。Source 元件可以處理各種型別、各種格式的日誌資料，包括 avro、thrift、exec、jms、spooling directory、netcat、taildir、sequence generator、syslog、http、legacy。

3 Sink

Sink 不斷地輪詢 Channel 中的事件且批量地移除它們，並將這些事件批量寫入到儲存或索引系統、或者被髮送到另一個 Flume Agent。Sink 元件目的地包括 hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定義。

4 Channel

Channel 是位於 Source 和 Sink 之間的緩衝區。因此，Channel 允許 Source 和 Sink 運作在不同的速率上。Channel 是執行緒安全的，可以同時處理幾個 Source 的寫入操作和幾個Sink 的讀取操作。Flume 自帶兩種 Channel：Memory Channel 和 File Channel。Memory Channel 是記憶體中的佇列。MemoryChannel 在不需要關心資料丟失的情景下適用。如果需要關心資料丟失，那麼 Memory Channel 就不應該使用，因為程式死亡、機器宕機或者重啟都會導致資料丟失。File Channel 將所有事件寫到磁碟。因此在程式關閉或機器宕機的情況下不會丟失資料。

5 Event

傳輸單元，Flume 資料傳輸的基本單元，以 Event 的形式將資料從源頭送至目的地。Event 由 Header 和 Body 兩部分組成，Header 用來存放該 event 的一些屬性，為K-V 結構，Body 用來存放該條資料，形式為位元組陣列。

入門

以官網的例子來看：Welcome to Apache Flume — Apache Flume 需要注意配置時，官網的加粗配置是必須有的

1.flume/job，在此目錄下建立flume-netcat-logger.conf，新增如下內容

# example.conf: A single-node Flume configuration

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

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

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

# Use a channel which buffers events in memory
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

　　　　2.啟動，分為兩種方式

第一種：

[[email protected] flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf - Dflume.root.logger=INFO,console

第二種：

[[email protected]]$ bin/flume-ng agent -c conf/ -n a1 -f job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console

引數說明：

--conf/-c：表示配置檔案儲存在 conf/目錄

--name/-n：表示給 agent 起名為 a1

--conf-file/-f：flume 本次啟動讀取的配置檔案是在 job 資料夾下的 flume-telnet.conf檔案。

-Dflume.root.logger=INFO,console ：-D 表示 flume 執行時動態修改 flume.root.logger

引數屬性值，並將控制檯日誌列印級別設定為 INFO 級別。日誌級別包括:log、info、warn、error。

　　　　3.使用 netcat 工具向本機的 44444 埠傳送內容

　　4.在Flume 監聽頁面觀察接收資料情況

　　　5.由此我們可以看出Flume的運作規律以及方式，結構也並不複雜

　　入門1.1

1)建立配置檔案flume-taildir-hdfs.conf
2)設定source為taildir
3)設定監控資料目錄自定義
4)設定channel為記憶體模式
5)設定sink為hdfs
6)上傳檔案字首設定為upload-
7)設定時間滾動資料夾為true
8)設定1小時建立一個新的資料夾
9)啟動本地時間戳
10)積攢100個event才flush到hdfs一次
11)設定檔案型別，可支援壓縮
12)設定60s生成一個新的檔案
13)設定每個檔案的滾動大小為128M
14)設定檔案的滾動和event的 數量無關
15)啟動flume
16)往監聽目錄下插入資料
17)檢視hdfs是否有資料

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

# Describe/configure the source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/log/a.log
a1.sources.r1.headers.f1.headerKey1 = value1

a1.sources.r1.fileHeader = true
a1.sources.ri.maxBatchCount = 1000

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path=/flume/%Y%m%d
a1.sinks.k1.hdfs.filePrefix = upload-

a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 1
a1.sinks.k1.hdfs.roundUnit = hour

a1.sinks.k1.hdfs.useLocalTimeStamp =true
a1.sinks.k1.hdfs.batchSize= 100

a1.sinks.k1.hdfs.codeC = gzip
a1.sinks.k1.hdfs.fileType = CompressedStream

a1.sinks.k1.hdfs.rollInterval = 60
a1.sinks.k1.hdfs.rollSize= 131072
a1.sinks.k1.hdfs.rollCount=0

# Use a channel which buffers events in memory
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

執行

bin/flume-ng agent --conf conf --conf-file job/flume-taildir-hdfs.conf  --name a1 -Dflume.root.logger=INFO,console

效果

Flume 相關學習

基礎框架

入門

4.在Flume 監聽頁面觀察接收資料情況

入門1.1

相關推薦

　　4.在Flume 監聽頁面觀察接收資料情況

　　入門1.1