作者:taozong2602912833 | 来源:互联网 | 2023-09-23 15:51
一:channel选择器Channel选择器用来解决source接收数据后写入到哪些channel,channel选择器的配置是通过channel处理器完成的,通过在配置文件中
一:channel选择器
Channel选择器用来解决source接收数据后写入到哪些channel,channel选择器的配置是通过channel处理器完成的,通过在配置文件中使用seletor后缀设置,flume内置了2种channel选择器,一种是复制,一种是分发,默认情况下,如果不特别指定,则默认是复制的形式,在设置选择器的时候,我们也可以设置哪些是必须,哪些是可选的,通过optional来进行设置。
二:复制channel选择器
? 在设置type的时候,type的值为replication,就代表复制,它有一个参数optional,这个参数是可选的,对于optional指定的channel,如果写入数据失败,Source不会抛出异常,相反,对于必须的channel,一旦channle内存满了,或者其它原因导致不能写入,Source将会抛出异常,然后重试。
??新建一个文件,配置如下:
a3.sources = r1
a3.sinks = k1 k2 k3
a3.channels = c1 c2 c3
a3.sources.r1.type=spooldir
a3.sources.r1.spoolDir = /flume/apache-flume-1.6.0-bin/replicat
a3.sources.r1.fileHeader = true
a3.sources.r1.channels = c1 c2 c3
a3.sources.r1.selector.optiOnal= c3
a3.sources.r1.selector.type=replicating
a3.sinks.k1.type = avro
a3.sinks.k1.channel = c1
a3.sinks.k1.hostname = ip
a3.sinks.k1.port = 4444
a3.sinks.k2.type = avro
a3.sinks.k2.channel = c2
a3.sinks.k2.hostname = ip
a3.sinks.k2.port = 5555
a3.sinks.k3.type =avro
a3.sinks.k3.channel = c3
a3.sinks.k3.hostname = ip
a3.sinks.k3.port = 6666
a3.channels.c1.type = memory
a3.channels.c1.capacity=1000
a3.channels.c1.transcatiOnCapacity=100
a3.channels.c2.type = memory
a3.channels.c2.capacity=1000
a3.channels.c2.transcatiOnCapacity=100
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transcatiOnCapacity=100
? 同一个数据源经不同的channel,写入不同的sink,其中c3是可选的,意思是如果写入c3失败,该失败会被无视,当我们往监控目录放入文件后,agent通过复制的方式,将内容分发到3个不同的channel,供sink读取,效果如下:
三:分发选择器
分发选择器我们可以理解为,可以根据不同的报文头值进行不同的路由,指定哪些事件可以进入哪些channel。
配置参数如下:
Type:multiplexing
Header:用来检查该事件路由到那个channel
Mapping.*?:?报文头映射的列表
Optional?:?可选的参数,如果写入失败,不会报错
新建一个文件,配置如下:
a0.sources = r1
a0.sinks = k1 k2 k3 k4
a0.channels = c1 c2 c3 c4
a0.sources.r1.type=exec
a0.sources.r1.command= cat /flume/apache-flume-1.6.0-bin/testFlumeDir/exec.log
a0.sources.r1.host = ip
a0.sources.r1.port = 8888
a0.sources.r1.channels = c1 c2 c3 c4
a0.sources.r1.interceptors = i1
a0.sources.r1.interceptors.i1.type = static
a0.sources.r1.interceptors.i1.key = airline
a0.sources.r1.interceptors.i1.value = CA
a0.sources.r1.selector.type=multiplexing
# airline的值CA,CZ
a0.sources.r1.selector.header = airline
a0.sources.r1.selector.mapping.CA = c1 c2
a0.sources.r1.selector.mapping.CZ = c2
a0.sources.r1.selector.optional.CA = c3
a0.sources.r1.selector.optional.CZ = c4
a0.sources.r1.selector.mapping.default = c4
a0.sinks.k1.type = avro
a0.sinks.k1.channel = c1
a0.sinks.k1.hostname = ip
a0.sinks.k1.port = 4444
a0.sinks.k2.type = avro
a0.sinks.k2.channel = c2
a0.sinks.k2.hostname = ip
a0.sinks.k2.port = 5555
a0.sinks.k3.type = avro
a0.sinks.k3.channel = c3
a0.sinks.k3.hostname = ip
a0.sinks.k3.port = 6666
a0.sinks.k4.type = avro
a0.sinks.k4.channel = c4
a0.sinks.k4.hostname = ip
a3.sinks.k4.port = 7777
a0.channels.c1.type = memory
a0.channels.c1.capacity=1000
a0.channels.c1.transcatiOnCapacity=100
a0.channels.c2.type = memory
a0.channels.c2.capacity=1000
a0.channels.c2.transcatiOnCapacity=100
a0.channels.c3.type = memory
a0.channels.c3.capacity=1000
a0.channels.c3.transcatiOnCapacity=100
a0.channels.c4.type = memory
a0.channels.c4.capacity=1000
a0.channels.c4.transcatiOnCapacity=10
配置中通过静态拦截器自定义报文头值,对于报文头里面key为airline,value为CA的会写入到C1,C2,C3这3个channel,且C3是可选的,即使写入C3失败,Source也不会抛出异常,不会重试。效果如下:
C1:
C2:
C3:
C4没收到数据: