Flume：
=====================
Flume是一種分布式的、可靠的、可用的服務，可以有效地收集、聚合和移動大量的日志數據。
它有一個基於流數據的簡單而靈活的體系結構。
它具有健壯性和容錯能力，具有可調的可靠性機制和許多故障轉移和恢復機制。
它使用一個簡單的可擴展數據模型，允許在線分析應用程序。
source：源
對channel而言，相當於生產者，通過接收各種格式數據發送給channel進行傳輸
channel：通道
相當於數據緩沖區，接收source數據發送給sink
sink：沉槽
對channel而言，相當於消費者，通過接收channel數據通過指定數據類型發送到指定位置
Event：
===============
flume傳輸基本單位：
head + body
flume安裝：
================
1、解壓
2、符號鏈接
3、配置環境變量並使其生效
4、修改配置文件
1)重命名flume-env.ps1.template為flume-env.ps1
2)重命名flume-env.sh.template為flume-env.sh
3)修改flume-env.sh，配置jdk目錄,添加
export JAVA_HOME=/soft/jdk
5、flume 查看版本
flume-ng version
flume使用：
=========================
//flume可以將配置文件寫在zk上
//flume運行命令
flume-ng agent -n a1 -f xxx.conf /flume-ng agent -n xx -f xxx.conf
agent: a1
source: s1
channel:c1
sink: n1
使用方法：
1、編寫配置文件r_nc.conf
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2、啟動flume，指定配置文件
flume-ng agent -n a1 -f r_nc.conf
3、啟動另一個會話，進行測試
nc localhost 8888
//用戶手冊
http://flume.apache.org/FlumeUserGuide.html
后台運行程序：
=============================================
ctrl + z :將程序放在后台運行 =====> [1]+ Stopped flume-ng agent -n a1 -f r_nc.conf
通過 bg %1 的方式將程序后台運行
通過jobs查看后台任務
通過 fg %1 的方式將程序放在前台運行
flume:
海量日志數據的收集、聚合和移動
flume-ng agent -n a1 -f xxx.conf
source
相對於channel是生產者 //netcat
channel
類似於緩沖區 //memory
sink
相對於channel是消費者 //logger
Event:
header + body
k v data
source:
============================================
1、序列(seq)源：多用作測試
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = seq
# 總共發送的事件個數
a1.sources.r1.totalEvents = 1000
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2、壓力(stress)源:多用作負載測試
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = org.apache.flume.source.StressSource
# 單個事件大小,單位：byte
a1.sources.r1.size = 10240
# 事件總數
a1.sources.r1.maxTotalEvents = 1000000
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
3、滾動目錄(Spooldir)源：監聽指定目錄新文件產生，並將新文件數據作為event發送
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = spooldir
# 設置監聽目錄
a1.sources.r1.spoolDir = /home/centos/spooldir
# 通過以下配置指定消費完成后文件后綴
#a1.sources.r1.fileSuffix = .COMPLETED
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
4、exec源 //通過執行linux命令產生新數據
//典型應用 tail -F (監聽一個文件,文件增長的時候，輸出追加數據)
//不能保證數據完整性，很可能丟失數據
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = exec
# 配置linux命令
a1.sources.r1.command = tail -F /home/centos/readme.txt
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
5、Taildir源 //監控目錄下文件
//文件類型可通過正則指定
//有容災機制
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = TAILDIR
# 設置source組 可設置多個
a1.sources.r1.filegroups = f1
# 設置組員的監控目錄和監控文件類型,使用正則表示,只能監控文件
a1.sources.r1.filegroups.f1 = /home/centos/taildir/.*
# 設置定位文件的位置
# a1.sources.r1.positionFile ~/.flume/taildir_position.json
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
sink：
====================================
1、fileSink //多用作數據收集
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = file_roll
# 配置目標文件夾
a1.sinks.k1.sink.directory = /home/centos/file
# 設置滾動間隔，默認30s，設為0則不滾動，成為單個文件
a1.sinks.k1.sink.rollInterval = 0
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2、hdfsSink //默認以seqFile格式寫入
//k：LongWritable
//v: BytesWritable
//
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = hdfs
# 配置目標文件夾
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/
# 配置文件前綴
a1.sinks.k1.hdfs.filePrefix = events-
# 滾動間隔，秒
a1.sinks.k1.hdfs.rollInterval = 0
# 觸發滾動文件大小，byte
a1.sinks.k1.hdfs.rollSize = 1024
# 配置使用本地時間戳
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 配置輸出文件類型，默認SequenceFile
# DataStream文本格式,不能設置壓縮編解碼器
# CompressedStream壓縮文本格式，需要設置編解碼器
a1.sinks.k1.hdfs.fileType = DataStream
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
3、hiveSink： //hiveserver幫助：hive --service help
//1、hive --service metastore 啟動hive的metastore服務，metastore地址：thrift://localhost:9083
//2、將hcatalog的依賴放在/hive/lib下，cp hive-hcatalog* /soft/hive/lib (位置/soft/hive/hcatalog/share/hcatalog)
//3、創建hive事務表
//SET hive.support.concurrency=true;
SET hive.enforce.bucketing=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
SET hive.compactor.initiator.on=true;
SET hive.compactor.worker.threads=1;
//create table myhive.weblogs(id int, name string, age int)
clustered by(id) into 2 buckets
row format delimited
fields terminated by '\t'
stored as orc
tblproperties('transactional'='true');
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = hive
a1.sinks.k1.hive.metastore = thrift://127.0.0.1:9083
a1.sinks.k1.hive.database = myhive
a1.sinks.k1.hive.table = weblogs
a1.sinks.k1.useLocalTimeStamp = true
#輸入格式，DELIMITED和json
#DELIMITED 普通文本
#json json文件
a1.sinks.k1.serializer = DELIMITED
#輸入字段分隔符,雙引號
a1.sinks.k1.serializer.delimiter = ","
#輸出字段分隔符,單引號
a1.sinks.k1.serializer.serdeSeparator = '\t'
#字段名稱，","分隔，不能有空格
a1.sinks.k1.serializer.fieldnames =id,name,age
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
4、hbaseSink //SimpleHbaseEventSerializer將rowKey和col設置了默認值，不能自定義
//RegexHbaseEventSerializer可以手動指定rowKey和col字段名稱
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = hbase
a1.sinks.k1.table = flume_hbase
a1.sinks.k1.columnFamily = f1
a1.sinks.k1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer
# 配置col正則手動指定
# rowKeyIndex手動指定rowKey，索引以0開頭
a1.sinks.k1.serializer.colNames = ROW_KEY,name,age
a1.sinks.k1.serializer.regex = (.*),(.*),(.*)
a1.sinks.k1.serializer.rowKeyIndex=0
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
5、asynchbaseSink //異步hbaseSink
//異步機制，寫入速度快
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = asynchbase
a1.sinks.k1.table = flume_hbase
a1.sinks.k1.columnFamily = f1
a1.sinks.k1.serializer = org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
channel：緩沖區
=====================================
1、memorychannel
a1.channels.c1.type = memory
# 緩沖區中存留的最大event個數
a1.channels.c1.capacity = 1000
# channel從source中每個事務提取的最大event數
# channel發送給sink每個事務發送的最大event數
a1.channels.c1.transactionCapacity = 100
2、fileChannel： //檢查點和數據存儲在默認位置時，當多個channel同時開啟
//會導致文件沖突，引發其他channel會崩潰
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels = c1
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/centos/flume/checkpoint
a1.channels.c1.dataDirs = /home/centos/flume/data
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
memoryChannel：快速，但是當設備斷電，數據會丟失
FileChannel： 速度較慢，即使設備斷電，數據也不會丟失
Avro
===============================================
source
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4444
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
***********************************************************************************************
*啟動avro客戶端，發送數據： *
* flume-ng avro-client -H localhost -p 4444 -R ~/avro/header.txt -F ~/avro/user0.txt *
* 指定ip 指定端口 指定header文件 指定數據文件 *
***********************************************************************************************
sink
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /home/centos/taildir/.*
# 配置sink
a1.sinks.k1.type = avro
a1.sinks.k1.bind = 192.168.23.101
a1.sinks.k1.port = 4444
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Flume躍點：
=====================================
1、將s101的flume發送到其他節點
xsync.sh /soft/flume
xsync.sh /soft/apache-flume-1.8.0-bin/
2、切換到root用戶，分發環境變量文件
su root
xsync.sh /etc/profile
exit
3、配置文件
1)配置s101 //hop.conf
設置source:avro
設置sink： hdfs
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4444
# 配置sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/hop/%y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollSize = 1024
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.fileType = DataStream
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2)配置s102-s104 //hop2.conf
設置source:taildir
設置sink： avro
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /home/centos/taildir/.*
# 配置sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.23.101
a1.sinks.k1.port = 4444
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
4、在s102-s104創建~/taildir文件夾
xcall.sh "mkdir ~/taildir"
5、啟動s101的flume
flume-ng agent -n a1 -f /soft/flume/conf/hop.conf
6、分別啟動s102-s104的flume,並將其放在后台運行
flume-ng agent -n a1 -f /soft/flume/conf/hop2.conf &
7、進行測試，分別在s102-s104的taildir中創建數據，觀察hdfs數據情況
s102]$ echo 102 > taildir/1.txt
s103]$ echo 103 > taildir/1.txt
s104]$ echo 104 > taildir/1.txt
interceptor：攔截器
==================================
是source端組件：負責修改或刪除event
每個source可以配置多個攔截器 ===> interceptorChain
1、Timestamp Interceptor //時間戳攔截器 + header
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
# 給攔截器起名
a1.sources.r1.interceptors = i1
# 指定攔截器類型
a1.sources.r1.interceptors.i1.type = timestamp
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2、Static Interceptor //靜態攔截器 + header
3、Host Interceptor //主機攔截器 + header
4、設置攔截器鏈：
# 將agent組件起名
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
a1.sources.r1.interceptors = i1 i2 i3
a1.sources.r1.interceptors.i1.type = timestamp
a1.sources.r1.interceptors.i2.type = host
a1.sources.r1.interceptors.i3.type = static
a1.sources.r1.interceptors.i3.key = location
a1.sources.r1.interceptors.i3.value = NEW_YORK
# 配置sink
a1.sinks.k1.type = logger
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
channel selector：通道挑選器
====================================
是source端組件：負責將event發送到指定的channel，相當於分區
當一個source設置多個channel時，默認以副本形式向每個channel發送一個event拷貝
1、replication副本通道挑選器 //默認挑選器，source將所有channel發送event副本
//設置source x 1, channel x 3, sink x 3
// nc memory file
# 將agent組件起名
a1.sources = r1
a1.sinks = k1 k2 k3
a1.channels = c1 c2 c3
# 配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 8888
a1.sources.r1.selector.type = replicating
# 配置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
a1.channels.c3.type = memory
a1.channels.c3.capacity = 1000
a1.channels.c3.transactionCapacity = 100
# 配置sink
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /home/centos/file1
a1.sinks.k1.sink.rollInterval = 0
a1.sinks.k2.type = file_roll
a1.sinks.k2.sink.directory = /home/centos/file2
a1.sinks.k2.sink.rollInterval = 0
a1.sinks.k3.type = file_roll
a1.sinks.k3.sink.directory = /home/centos/file3
a1.sinks.k3.sink.rollInterval = 0
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1 c2 c3
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
a1.sinks.k3.channel = c3
2、Multiplexing 多路復用通道挑選器 //選擇avro源發送文件
# 將agent組件起名
a1.sources = r1
a1.sinks = k1 k2 k3
a1.channels = c1 c2 c3
# 配置source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4444
# 配置通道挑選器
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = country
a1.sources.r1.selector.mapping.CN = c1
a1.sources.r1.selector.mapping.US = c2
a1.sources.r1.selector.default = c3
# 配置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
a1.channels.c3.type = memory
a1.channels.c3.capacity = 1000
a1.channels.c3.transactionCapacity = 100
# 配置sink
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /home/centos/file1
a1.sinks.k1.sink.rollInterval = 0
a1.sinks.k2.type = file_roll
a1.sinks.k2.sink.directory = /home/centos/file2
a1.sinks.k2.sink.rollInterval = 0
a1.sinks.k3.type = file_roll
a1.sinks.k3.sink.directory = /home/centos/file3
a1.sinks.k3.sink.rollInterval = 0
# 綁定channel-source, channel-sink
a1.sources.r1.channels = c1 c2 c3
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
a1.sinks.k3.channel = c3
1、創建file1 file2 file3文件夾，家目錄
mkdir file1 file2 file3
2、創建文件夾country，並放入頭文件和數據
創建頭文件CN.txt、US.txt、OTHER.txt
CN.txt ===> country CN
US.txt ===> country US
OTHER.txt ===> country OTHER
創建數據 1.txt
1.txt ====> helloworld
3、運行flume
flume-ng agent -n a1 -f /soft/flume/selector_multi.conf
4、運行Avro客戶端
flume-ng avro-client -H localhost -p 4444 -R ~/country/US.txt -F ~/country/1.txt ===> 查看file2
flume-ng avro-client -H localhost -p 4444 -R ~/country/CN.txt -F ~/country/1.txt ===> 查看file1
flume-ng avro-client -H localhost -p 4444 -R ~/country/OTHER.txt -F ~/country/1.txt ===> 查看file3
sinkProcessor
=================================
sink Runner 運行一個 sink Group
sink Group 是由一個或多個 sink 構成
sink Runner 告訴 sink Group 處理下一批 event
sink Group 含有一個 sink Processor , 負責指定一個 sink 來處理這批數據
2、failover 容災 //將所有sink設置一個優先級
//數量越大，優先級越高
//當數據傳入時，優先級最高的sink負責處理
//當sink掛掉，次高優先級的sink被激活，繼續處理數據
//channel和sink必須一對一
a1.sources = r1
a1.sinks = s1 s2 s3
a1.channels = c1 c2 c3
# Describe/configure the source
a1.sources.r1.type = seq
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = s1 s2 s3
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.s1 = 5
a1.sinkgroups.g1.processor.priority.s2 = 10
a1.sinkgroups.g1.processor.priority.s3 = 15
a1.sinkgroups.g1.processor.maxpenalty = 10000
# Describe the sink
a1.sinks.s1.type = file_roll
a1.sinks.s1.sink.directory = /home/centos/file1
a1.sinks.s2.type = file_roll
a1.sinks.s2.sink.directory = /home/centos/file2
a1.sinks.s3.type = file_roll
a1.sinks.s3.sink.directory = /home/centos/file3
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c2.type = memory
a1.channels.c3.type = memory
# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2 c3
a1.sinks.s1.channel = c1
a1.sinks.s2.channel = c2
a1.sinks.s3.channel = c3
Event事件是由Source端封裝輸入數據的字節數組得來的
Event event = EventBuilder.withBody(body);
Sink中的process方法返回兩種狀態：
1、READY //一個或多個event成功分發
2、BACKOFF //channel中沒有數據提供給sink
flume中事務的生命周期：
tx.begin() //開啟事務，之后執行操作
tx.commit() //提交事務，操作完成后由此提交
tx.rollback() //回滾事務，出現異常可以采取回滾措施
tx.close() //關閉事務，最后一定要關閉事務