flume 的source 、channel和sink 多种组合

flume 有三大组件source 、channel和sink,各个组件之间都可以相互组合使用,各组件间耦合度低。使用灵活,方便。

1.多sink

channel 的内容只输出一次,同一个event 如果sink1 输出,sink2 不输出;如果sink1 输出,sink1 不输出。 最终 sink1+sink2=channel 中的数据。

配置文件如下:

a1.sources=r1a1.sinks= k1 k2a1.channels= c1# Describe/configure the sourcea1.sources.r1.type= execa1.sources.r1.shell= /bin/bash -ca1.sources.r1.channels= c1a1.sources.r1.command= tail -F /opt/apps/logs/tail4.log# channela1.channels.c1.type= memorya1.channels.c1.capacity=1000a1.channels.c1.transactionCapacity=100#sink1a1.sinks.k1.channel= c1a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.topic= mytopica1.sinks.k1.kafka.bootstrap.servers= localhost:9092a1.sinks.k1.kafka.flumeBatchSize=20a1.sinks.k1.kafka.producer.acks=1a1.sinks.k1.kafka.producer.linger.ms=1a1.sinks.ki.kafka.producer.compression.type= snappy#sink2a1.sinks.k2.type= file_rolla1.sinks.k2.channel= c1#a1.sinks.k2.sink.rollInterval=0a1.sinks.k2.sink.directory= /opt/apps/tmp

2.多 channel 多sink ,每个sink 输出内容一致

(memory channel 用于kafka操作,实时性高,file channel 用于 sink file 数据安全性高) 

(多channel 单 sink 的情况没有举例,个人感觉用处不广泛。)

配置文件如下:

a1.sources=r1a1.sinks= k1 k2a1.channels= c1 c2# Describe/configure the sourcea1.sources.r1.type= execa1.sources.r1.shell= /bin/bash -ca1.sources.r1.channels= c1 c2a1.sources.r1.command= tail -F /opt/apps/logs/tail4.log#多个channel 的数据相同a1.sources.r1.selector.type=replicating# channel1a1.channels.c1.type= memorya1.channels.c1.capacity=1000a1.channels.c1.transactionCapacity=100#channel2a1.channels.c2.type= filea1.channels.c2.checkpointDir= /opt/apps/flume-1.7.0/checkpointa1.channels.c2.dataDirs= /opt/apps/flume-1.7.0/data#sink1a1.sinks.k1.channel= c1a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.topic= mytopica1.sinks.k1.kafka.bootstrap.servers= localhost:9092a1.sinks.k1.kafka.flumeBatchSize=20a1.sinks.k1.kafka.producer.acks=1a1.sinks.k1.kafka.producer.linger.ms=1a1.sinks.ki.kafka.producer.compression.type= snappy#sink2a1.sinks.k2.type= file_rolla1.sinks.k2.channel= c2#a1.sinks.k2.sink.rollInterval=0a1.sinks.k2.sink.directory= /opt/apps/tmp

3. 多source 单 channel 单 sink

多个source 可以读取多种信息放在一个channel 然后输出到同一个地方 

配置文件如下:

a1.sources=r1r2a1.sinks= k1a1.channels= c1# source1a1.sources.r1.type= execa1.sources.r1.shell= /bin/bash -ca1.sources.r1.channels= c1a1.sources.r1.command= tail -F /opt/apps/logs/tail4.log# source2a1.sources.r2.type= execa1.sources.r2.shell= /bin/bash -ca1.sources.r2.channels= c1a1.sources.r2.command= tail -F /opt/apps/logs/tail2.log# channel1  in memorya1.channels.c1.type= memorya1.channels.c1.capacity=1000a1.channels.c1.transactionCapacity=100#sink1a1.sinks.k1.channel= c1a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.topic= mytopica1.sinks.k1.kafka.bootstrap.servers= localhost:9092a1.sinks.k1.kafka.flumeBatchSize=20a1.sinks.k1.kafka.producer.acks=1a1.sinks.k1.kafka.producer.linger.ms=1a1.sinks.ki.kafka.producer.compression.type= snappy

flume 像乐高积木一样可以自己随心所欲将不同的组件进行搭配使用,耦合度低。



Source

rpc远程过程调用协议,客户机与服务机的调用模式需要对数据进行序列化。

         1:客户机将参数序列化并以二进制形式通过网络传输到服务器。

         2:服务器接收到后进行反序列化再调用方法获取返回值。

         3:服务器将返回值序列化后再通过网络传输给客户机。

         4:客户机接收到结果后再进行反序列化获取结果。

Avro source:

         Avro就是一种序列化形式,avrosource监听一个端口只接收avro序列化后的数据,其他类型的不接收。

         type:avrosource的类型,必须是avro。

bind:要监听的(本机的)主机名或者ip。此监听不是过滤发送方。一台电脑不是说只有一个IP。有多网卡的电脑,对应多个IP。

port:绑定的本地的端口。


Thrif source:

         和avro一样是一种数据序列化形式,Thrifsource只采集thrift数据序列化后的数据


Exec source:

         采集linux命令的返回结果传输给channel

         type:source的类型:必须是exec。

command:要执行命令。

tail  –f  若文件被删除即使重新创建同名文件也不会监听

        tail  -F  只要文件同名就可以继续监听

以上可以用在日志文件切割时的监听



JMS Source:

Java消息服务数据源,Java消息服务是一个与具体平台无关的API,这是支持jms规范的数据源采集;


Spooling Directory Source:通过文件夹里的新增的文件作为数据源的采集;


Kafka Source:从kafka服务中采集数据。


NetCat Source:绑定的端口(tcp、udp),将流经端口的每一个文本行数据作为Event输入

        type:source的类型,必须是netcat。

bind:要监听的(本机的)主机名或者ip。此监听不是过滤发送方。一台电脑不是说只有一个IP。有多网卡的电脑,对应多个IP。

port:绑定的本地的端口。


HTTP Source:监听HTTP POST和 GET产生的数据的采集


Chanel

         是一个数据存储池,中间通道,从source中接收数据再向sink目的地传输,如果sink写入失败会自动重写因此不会造成数据丢失。

         Memory:用内存存储,但服务器宕机会丢失数据。

                 Typechannel的类型:必须为memory

capacity:channel中的最大event数目

transactionCapacity:channel中允许事务的最大event数目


         File:使用文件存储数据不会丢失数据但会耗费io。

                 Typechannel的类型:必须为 file

checkpointDir :检查点的数据存储目录

dataDirs :数据的存储目录

transactionCapacity:channel中允许事务的最大event数目


         SpillableMemory Channel:内存文件综合使用,先存入内存达到阀值后flush到文件中。

                Typechannel的类型:必须为SPILLABLEMEMORY

memoryCapacity:内存的容量event数

overflowCapacity:数据存到文件的event阀值数

checkpointDir:检查点的数据存储目录

dataDirs:数据的存储目录


         Jdbc:使用jdbc数据源来存储数据。

         Kafka:使用kafka服务来存储数据。


Sink

         各种类型的目的地,接收channel写入的数据并以指定的形式表现出来。Sink有很多种类型。

type:sink的类型 必须是hdfs。

hdfs.path:hdfs的上传路径。

hdfs.filePrefix:hdfs文件的前缀。默认是:FlumeData

hdfs.rollInterval:间隔多久产生新文件,默认是:30(秒) 0表示不以时间间隔为准。

hdfs.rollSize:文件到达多大再产生一个新文件,默认是:1024(bytes)0表示不以文件大小为准。

hdfs.rollCount:event达到多大再产生一个新文件,默认是:10(个)0表示不以event数目为准。

hdfs.batchSize:每次往hdfs里提交多少个event,默认为100

hdfs.fileType:hdfs文件的格式主要包括:SequenceFile,DataStream ,CompressedStream,如果使用了CompressedStream就要设置压缩方式。

hdfs.codeC:压缩方式:gzip,bzip2, lzo, lzop, snappy

注:%{host}可以使用header的key。以及%Y%m%d来表示时间,但关于时间的表示需要在header里有timestamp这个key。


Logger Sink将数据作为日志处理(根据flume中的设置的日志方式来显示)

要在控制台显示在运行agent的时候加入:-Dflume.root.logger=INFO,console。

type:sink的类型:必须是logger。

maxBytesToLog:打印body的最长的字节数 默认为16



Avro Sink:数据被转换成Avro Event,然后发送到指定的服务端口上。

                 type:sink的类型:必须是 avro。

hostname:指定发送数据的主机名或者ip

port:指定发送数据的端口

实例

1:监听一个文件的增加变化,采集数据并在控制台打印。

在这个例子中我使用exec source,memory chanel,logger sink。可以看我的agent结构图

以下是我创建的exec_source.conf

a1.sources=r1

a1.channels=c1

a1.sinks=k1


a1.sources.r1.type=exec

a1.sources.r1.command=tail -F/usr/local/success.log


a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactioncapacity=100


a1.sinks.k1.type=logger


a1.sources.r1.channels=c1

a1.sinks.k1.channel=c1


执行命令:

bin/flume-ngagent --conf conf/ --conf-file conf/exec_source.conf --name a1-Dflume.root.logger=INFO,console &


然后更改/usr/local/success.log文件中的内容后可以看到flume采集到了文件的变化并在控制台上打印出来。文件初始内容hello和how are you,剩下的i am fine和ok为新增加内容。

2:监控一个文件变化并将其发送到另一个服务器上然后打印

这个例子可以建立在上一个例子之上,但是需要对flume的结构做一些修改,我使用avro序列化数据再发送到指定的服务器上。详情看结构图。

实际上flume可以进行多个节点关联,本例中我只使用131向139发送数据

131,139上都必须启动agent

服务器131配置

以下是我创建的exec_source_avro_sink.conf

a1.sources=r1

a1.channels=c1

a1.sinks=k1


a1.sources.r1.type=exec

a1.sources.r1.command=tail -F/usr/local/success.log


a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactioncapacity=100


a1.sinks.k1.type=avro

a1.sinks.k1.hostname=192.168.79.139

a1.sinks.k1.port=42424


a1.sources.r1.channels=c1

a1.sinks.k1.channel=c1


执行命令启动agent

bin/flume-ng agent --conf conf/ --conf-fileconf/exec_source_avro_sink.conf --name a1 -Dflume.root.logger=INFO,console&


139服务器配置

执行命令拷贝flume到139

scp -r apache-flume-1.7.0-bin/root@192.168.79.139:/usr/local/

修改exec_source_avro_sink.conf

a1.sources=r1

a1.channels=c1

a1.sinks=k1


a1.sources.r1.type=avro

a1.sources.r1.bind=0.0.0.0

a1.sources.r1.port=42424


a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactioncapacity=100


a1.sinks.k1.type=logger

a1.sources.r1.channels=c1

a1.sinks.k1.channel=c1


执行命令启动agent

bin/flume-ng agent --conf conf/ --conf-fileconf/exec_source_avro_sink.conf --name a1 -Dflume.root.logger=INFO,console&


结果可以在139控制台上看到131中修改success.log的变化信息

3:avro-client实例

执行bin/flume-ng会提示有命令如下

help                     display this help text

agent                     run aFlume agent

avro-client               run anavro Flume client

version                   show Flume version info


avro-clinet是avro客户端,可以把本地文件以avro序列化方式序列化后发送到指定的服务器端口。本例就是将131的一个文件一次性的发送到139中并打印。

Agent结构图如下

131启动的是一个avro-client,它会建立连接,发送数据,断开连接,它只是一个客户端。

启动一个avro客户端

bin/flume-ngavro-client --conf conf/ --host 192.168.79.139 --port 42424 --filename/usr/local/success.log --headerFile /usr/local/kv.log


--headerFile是用来区分是哪个服务器发送的数据,kv.log中的内容会被发送到139,可以作为标识来使用。


139的avro_client.conf如下

a1.sources=r1

a1.channels=c1

a1.sinks=k1


a1.sources.r1.type=avro

a1.sources.r1.bind=0.0.0.0

a1.sources.r1.port=42424



a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactioncapacity=100


a1.sinks.k1.type=logger


a1.sources.r1.channels=c1

a1.sinks.k1.channel=c1


启动agent

bin/flume-ngagent --conf conf/ --conf-file conf/avro_client.conf --name a1-Dflume.root.logger=INFO,console &


139控制台显示如下

可以看到headers的内容headers:{hostname=192.168.79.131}

注意:

1:Flume服务没有stop命令需要通过kill来杀掉进行,可以使用jps  -m来确认是那个agent的number

[root@shb01 conf]# jps -m

3610 Jps -m

3512 Application --conf-fileconf/exec_source.conf --name a1


2:修改flume的配置文件后如avro_client.conf,flume会自动重启


3:logger sink默认只显示16个字节


4:flume是以event为单位进行数据传输的,其中headers是一个map容器map

Event: { headers:{hostname=192.168.79.131}body: 31 61                                           1a }


5:flume支持多节点关联但是sink和source的类型要一致,比如avro-client发送数据那么接收方的source也必须是avro否则会警告。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 195,719评论 5 462
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 82,337评论 2 373
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 142,887评论 0 324
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 52,488评论 1 266
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 61,313评论 4 357
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 46,284评论 1 273
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 36,672评论 3 386
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 35,346评论 0 254
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 39,644评论 1 293
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,700评论 2 312
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,457评论 1 326
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 32,316评论 3 313
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 37,706评论 3 299
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,990评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,261评论 1 251
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 41,648评论 2 342
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 40,859评论 2 335

推荐阅读更多精彩内容