Flume配置Failover Sink Processor

1 官网内容

2 看一张图一目了然

3 详细配置

  source配置文件

1#配置文件: 2 a1.sources= r1 3 a1.sinks= k1 k2 4 a1.channels= c1 5 6 #负载平衡 7 a1.sinkgroups = g1 8 a1.sinkgroups.g1.sinks = k1 k2 9 a1.sinkgroups.g1.processor.type = failover 10 a1.sinkgroups.g1.processor.priority.k1 = 5 11 a1.sinkgroups.g1.processor.priority.k2 = 10 12 a1.sinkgroups.g1.processor.maxpenalty = 1000 13 14 15 16 #Describe/configure the source 17 a1.sources.r1.type= exec 18 a1.sources.r1.command= tail -F /tmp/logs/test.log 19 20 21 #Describe the sink 22 a1.sinks.k1.type= avro 23 a1.sinks.k1.hostname= 127.0.0.1 24 a1.sinks.k1.port= 50001 25 26 a1.sinks.k2.type= avro 27 a1.sinks.k2.hostname= 127.0.0.1 28 a1.sinks.k2.port= 50002 29 30 # Usea channel which buffers events in memory 31 a1.channels.c1.type= memory 32 a1.channels.c1.capacity= 1000 33 a1.channels.c1.transactionCapacity= 100 34 35 # set channel 36 a1.sinks.k1.channel= c1 37 a1.sinks.k2.channel= c1 38 a1.sources.r1.channels= c1

  sink1配置文件

1# Name the components on this agent 2 a2.sources = r1 3 a2.sinks = k1 4 a2.channels = c1 5 6 # Describe/configure the source 7 a2.sources.r1.type = avro 8 a2.sources.r1.channels = c1 9 a2.sources.r1.bind = 127.0.0.1 10 a2.sources.r1.port = 50001 11 12 # Describe the sink 13 a2.sinks.k1.type = logger 14 a2.sinks.k1.channel = c1 15 16 # Use a channel which buffers events inmemory 17 a2.channels.c1.type = memory 18 a2.channels.c1.capacity = 1000 19 a2.channels.c1.transactionCapacity = 100

  sink2配置

1# Name the components on this agent 2 a3.sources = r1 3 a3.sinks = k1 4 a3.channels = c1 5 6 # Describe/configure the source 7 a3.sources.r1.type = avro 8 a3.sources.r1.channels = c1 9 a3.sources.r1.bind = 127.0.0.1 10 a3.sources.r1.port = 50002 11 12 # Describe the sink 13 a3.sinks.k1.type = logger 14 a3.sinks.k1.channel = c1 15 16 # Use a channel which buffers events inmemory 17 a3.channels.c1.type = memory 18 a3.channels.c1.capacity = 1000 19 a3.channels.c1.transactionCapacity = 100

4 启动服务

1先启动sink1 sink2 再启动source 2 3 flume-ng agent -c conf -f /mnt/software/flume-1.6.0/flume-conf/failOver/sink2.conf -n a3 -Dflume.root.logger=DEBUG,console 4 flume-ng agent -c conf -f /mnt/software/flume-1.6.0/flume-conf/failOver/sink1.conf -n a2 -Dflume.root.logger=DEBUG,console 5 flume-ng agent -c conf -f /mnt/software/flume-1.6.0/flume-conf/failOver/load_source_case.conf -n a1 -Dflume.root.logger=DEBUG,console

  5 效果测试

1启动后第一次走了sink2 2 3 : /127.0.0.1:42828 4 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 68 61 64 6F 6F 70 hadoop } 5 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 7A 68 61 6E 67 6A 69 6E zhangjin } 6 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 78 78 78 78 xxxx } 7 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 79 79 79 79 yyyy } 8 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 7A 68 61 6E 67 6A 69 6E zhangjin } 9 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 78 78 78 78 xxxx } 10 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 79 79 79 79 yyyy } 11 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 7A 68 61 6E 67 6A 69 6E zhangjin } 12 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 78 78 78 78 xxxx } 13 19/02/21 23:45:47 INFO sink.LoggerSink: Event: { headers:{} body: 79 79 79 79 yyyy } 14 15挂掉sink2,之后source感知到sink2挂了 16 17 Caused by: java.net.ConnectException: Connection refused 18 at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) 19 at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) 20 at org.jboss.netty.channel.socket.nio.NioClientBoss.connect(NioClientBoss.java:148) 21 at org.jboss.netty.channel.socket.nio.NioClientBoss.processSelectedKeys(NioClientBoss.java:104) 22 at org.jboss.netty.channel.socket.nio.NioClientBoss.process(NioClientBoss.java:78) 23 at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) 24 at org.jboss.netty.channel.socket.nio.NioClientBoss.run(NioClientBoss.java:41) 25 ... 3 more 26 27 28数据发往sink1 29 30 19/02/21 23:45:41 INFO ipc.NettyServer: [id: 0x77bfe0b5, /127.0.0.1:47142 => /127.0.0.1:50001] BOUND: /127.0.0.1:50001 31 19/02/21 23:45:41 INFO ipc.NettyServer: [id: 0x77bfe0b5, /127.0.0.1:47142 => /127.0.0.1:50001] CONNECTED: /127.0.0.1:47142 32 19/02/21 23:47:14 INFO sink.LoggerSink: Event: { headers:{} body: 7A 68 61 6E 67 6A 69 6E zhangjin } 33 19/02/21 23:47:14 INFO sink.LoggerSink: Event: { headers:{} body: 78 78 78 78 xxxx } 34 19/02/21 23:47:14 INFO sink.LoggerSink: Event: { headers:{} body: 79 79 79 79 yyyy }

  6 总结,从效果来看sink2挂了之后,数据发往sink1,实现了失败迁移的功能。

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

皕杰报表之UUID

​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段,为你的工作保驾护航

有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )