Flume概述

1)官網地址
http://flume.apache.org/

2）日誌採集工具

Flume是一種分散式，可靠且可用的服務，用於有效地收集，聚合和移動大量日誌資料。它具有基於
流資料流的簡單靈活的架構。它具有可靠的可靠性機制和許多故障轉移和恢復機制，具有強大的容錯
能力。它使用簡單的可擴充套件資料模型，允許線上分析應用程式。

3）為什麼需要flume

資料從哪裡來？
-》爬蟲
-》日誌資料 flume
-》傳統型資料庫 sqoop

4)flume架構

source:資料來源
產生資料流，同時source將產生的資料流傳輸到channel

channel:傳輸通道
用於橋接Source和sinks

sinks:下沉(資料池)
從channel收集資料

event:傳輸單元
Flume資料傳傳輸的基本單元，以事件的形式將資料送往目的地。

Flume安裝部署

1）下載安裝包
http://archive.apache.org/dist/flume/1.6.0/

2）上傳到linux
alt+p

3）解壓
tar -zxvf .tar

4)重新命名
mv flume-env.sh.template flume-env.sh

5)修改配置檔案
export JAVA_HOME=/opt/java/jdk1.8.0_141

場景一：使用Flume監聽一個埠

1）安裝telnet
yum search telnet
yum intsall telnet.x86_64

2)寫配置檔案
flumejob_telnet.conf

3)啟動
bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flumejob_telne
t.conf -Dflume.root.logger=INFO,console
(-Dflume.root.logger=INFO.console將監聽的日誌資訊直接輸出到控制檯上)
4)傳送資料
telnet localhost 44444

5)檢視

flumejob_telnet.conf:

#smple.conf: A single-node Flume configuration

# Name the components on this agent 定義變數方便呼叫 加s可以有多個此角色
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source 描述source角色 進行內容定製
# 此配置屬於tcp source 必須是netcat型別
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（file） 使用記憶體 總大小1000 每次傳輸100
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel 一個source可以繫結多個channel 
# 一個sinks可以只能繫結一個channel  使用的是圖二的模型
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

 

場景二:用Flume實時的採集檔案到HDFS

hive:
啟動
bin/flume-ng agent --conf conf/ --name a1 --confile conf/flumejob_hdfs.co
nf

配置資訊
flumejob_hdfs.conf ：

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

# Describe/configure the source 
# exec 執行一個命令的方式去檢視檔案 tail -F 實時檢視
a1.sources.r1.type = exec
# 要執行的指令碼command tail -F 預設10行 man tail  檢視幫助
a1.sources.r1.command = tail -F /tmp/root/hive.log
# 執行這個command使用的是哪個指令碼 -c 指定使用什麼命令
# whereis bash
# bash: /usr/bin/bash /usr/share/man/man1/bash.1.gz 
a1.sources.r1.shell = /usr/bin/bash -c

# Describe the sink 
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://master:9000/flume/%Y%m%d/%H
#上傳檔案的字首
a1.sinks.k1.hdfs.filePrefix = logs-
#是否按照時間滾動資料夾
a1.sinks.k1.hdfs.round = true
#多少時間單位建立一個新的資料夾  秒 （預設30s）
a1.sinks.k1.hdfs.roundValue = 1
#重新定義時間單位（每小時滾動一個資料夾）
a1.sinks.k1.hdfs.roundUnit = minute
#是否使用本地時間戳
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#積攢多少個 Event 才 flush 到 HDFS 一次
a1.sinks.k1.hdfs.batchSize = 500
#設定檔案型別，可支援壓縮
a1.sinks.k1.hdfs.fileType = DataStream
#多久生成一個新的檔案 秒
a1.sinks.k1.hdfs.rollInterval = 30
#設定每個檔案的滾動大小 位元組（最好128M）
a1.sinks.k1.hdfs.rollSize = 134217700
#檔案的滾動與 Event 數量無關
a1.sinks.k1.hdfs.rollCount = 0
#最小冗餘數(備份數 生成滾動功能則生效roll hadoop本身有此功能 無需配置) 1份 不冗餘
a1.sinks.k1.hdfs.minBlockReplicas = 1

# 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/ --name a1 --confile conf/flumejob_dir.con
f

flumejob_dir.conf ：

# 定義
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = spooldir
# 監控的資料夾
a1.sources.r1.spoolDir = /root/spooldir
# 上傳成功後顯示字尾名 
a1.sources.r1.fileSuffix = .COMPLETED
# 如論如何 加絕對路徑的檔名 預設false
a1.sources.r1.fileHeader = true

#忽略所有以.tmp 結尾的檔案（正在被寫入），不上傳
# ^以任何開頭 出現無限次 以.tmp結尾的
a1.sources.r1.ignorePattern = ([^ ]*\.tmp)

# Describe the sink 
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://master:9000/flume/spooldir/%Y%m%d/%H
#上傳檔案的字首
a1.sinks.k1.hdfs.filePrefix = spooldir-
#是否按照時間滾動資料夾
a1.sinks.k1.hdfs.round = true
#多少時間單位建立一個新的資料夾
a1.sinks.k1.hdfs.roundValue = 1
#重新定義時間單位
a1.sinks.k1.hdfs.roundUnit = hour
#是否使用本地時間戳
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#積攢多少個 Event 才 flush 到 HDFS 一次
a1.sinks.k1.hdfs.batchSize = 50

#設定檔案型別，可支援壓縮
a1.sinks.k1.hdfs.fileType = DataStream
#多久生成一個新的檔案
a1.sinks.k1.hdfs.rollInterval = 600
#設定每個檔案的滾動大小大概是 128M 
a1.sinks.k1.hdfs.rollSize = 134217700
#檔案的滾動與 Event 數量無關
a1.sinks.k1.hdfs.rollCount = 0
#最小副本數
a1.sinks.k1.hdfs.minBlockReplicas = 1

# 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