flume

一、

flume是分布式的，可靠的，用于从不同的来源有效收集 聚集 和 移动 大量的日志数据用以集中式的数据存储的系统。


二、

安装

tar -zxvf flume.tar.gz

三、source

1.测试

cd ./conf

vim template11.conf
＃命名Agent a1的组件
a1.sources  =  r1
a1.sinks  =  k1
a1.channels  =  c1

＃描述/配置Source
a1.sources.r1.type  =  netcat
a1.sources.r1.bind  =  0.0.0.0
a1.sources.r1.port  =  44444

＃描述Sink
a1.sinks.k1.type  =  logger

＃描述内存Channel
a1.channels.c1.type  =  memory
a1.channels.c1.capacity  =  1000
a1.channels.c1.transactionCapacity  =  100

＃为Channle绑定Source和Sink
a1.sources.r1.channels  =  c1
a1.sinks.k1.channel  =  c1


cd ../bin
./flume-ng agent --conf ../conf --conf-file ../conf/template1.conf --name a1 -Dflume.root.logger=INFO,console



客户端
telnet 192.168.76.131 44444



ctrl + ] 然后回车，输入的内容可见，否则，输入的内容不可见

输入内容：hello study ,回车 ，



服务端
有内容输出 接收到 hello study




2.监控指定目录

vim template2.conf

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


a1.sources.r1.type  = spooldir
a1.sources.r1.spoolDir  = /usr/local/software/flume/data


a1.sinks.k1.type  =  logger

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


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

若目录中有文件新增，则输入文件内容


五、channel

缓冲池，接收 source 传输的数据并缓存

六、sinks

取出 channel 中的数据

七、

描述：
向 机器 1 ，发送请求， 机器 1 传输给 2 ，2 传输给 3 ，3 在控制台输出

分析：
机器1
source : http ，由于发起的是 http 请求
channel : memoryChannel
sinks : avro

机器2
source : avro
channel : memoryChannel ，无特殊要求
sinks = avro

机器3
source : avro
channel : memoryChannel
sinks : logger 控制台打印

配置：
linux01
cd flume/conf
vim template3.conf


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

a1.sources.r1.type  = http
a1.sources.r1.port = 44444


a1.sinks.k1.type  =  avro
a1.sinks.k1.hostname = 192.168.76.134
a1.sinks.k1.port = 44444

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


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


配置linux02



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

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


a1.sinks.k1.type  =  avro
a1.sinks.k1.hostname = 192.168.76.132
a1.sinks.k1.port = 44444

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


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

配置 linux03

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

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


a1.sinks.k1.type  =  logger

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


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

依次启动服务

./flume-ng agent --conf ../conf --conf-file ../conf/template3.conf --name a1 -Dflume.root.logger=INFO,console


由依赖关系可知， linux01 需等待 linux02 ，linux02 需等待 linux03 ，
所以先从 linux03 开始启动

测试：
curl -X post -d '[{"headers":{"a":"a1","b":"b1"},"body":"hello-http-flume"}]' http://192.168.76.131:44444

结果：
linux 03 接收到 发送的数据

八、扇入实例

linux01 linux02 均向 linux03 发送数据

linux01
source: http
channel:memoryChannel
sinks = avro

linux02:同上

linux03:
source :avro
channel:memoryChannel
sinks = logger


九、扇出实例

linux01 向 linux02 linux03 发送数据

linux02 03
source : avro
channel : memoryChannel
sinks : logger

linux01
配置两个sinks 两个 channel

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


a1.sources.r1.type  = http
a1.sources.r1.port = 44444


a1.sinks.k1.type  =  avro
a1.sinks.k1.hostname = 192.168.76.132
a1.sinks.k1.port = 44444

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


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.sources.r1.channels  =  c1 c2
a1.sinks.k1.channel  =  c1
a1.sinks.k2.channel  =  c2

linux02 linux03

发送数据：
新开标签页，向linux01 发送数据
curl -X post -d '[{"headers":{"a":"a1","b":"b1"},"body":"hello-http-flume"}]' http://192.168.76.131:44444

结果：
linux02 linux03 均接收到数据

九、
hdfs sinks

配置：
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=hdfs://0.0.0.0:9000/ppp

启动 hdfs
cd /haddop/bin
start-dfs.sh

启动flume

curl -X post -d '[{"headers":{"a":"a1","b":"b1"},"body":"hello-http-flume"}]' http://192.168.76.131:44444

在
http://192.168.76.131:50070/explorer.html#/flume/data
访问查看传输的文件

十、selector 选择器

扇出时，数据时复制发送，即同时发给扇出接收方，数据时复制的，
现在判断将哪些数据发送给指定的接收方

a1.sources = r1
a1.channels = c1 c2 c3 c4
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = state
a1.sources.r1.selector.mapping.CZ = c1
a1.sources.r1.selector.mapping.US = c2 c3
a1.sources.r1.selector.default = c4
配置说明：
当 state = cz 时 ，发送给 缓冲池 c1 ，默认发送 缓冲池 c4
selector 配置在 source 上

配置：
linux01
a1.sources  =  r1
a1.sinks  =  k1 k2
a1.channels  =  c1 c2


a1.sources.r1.type  = http
a1.sources.r1.port = 44444
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = gender
a1.sources.r1.selector.mapping.male = c1
a1.sources.r1.selector.mapping.female = c2
a1.sources.r1.selector.default = c1


a1.sinks.k1.type  =  avro
a1.sinks.k1.hostname = 192.168.76.132
a1.sinks.k1.port = 44444

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

linux02 03 同扇出的配置


测试：
curl -X post -d '[{"headers":{"gender":"male"},"body":"hello-http-flume"}]' http://192.168.76.131:44444
只有