flume-1.6.0 高可用测试&&数据入Kafka

2024-11-26 技术教程

机器列表：

192.168.137.115slave0（agent）192.168.137.116slave1（agent）192.168.137.117slave2（agent）192.168.137.118slave3（collector）192.168.137.119slave4（collector）

在每个机器上创建目录

mkdir -p /home/qun/data/flume/logs

mkdir -p /home/qun/data/flume/data

mkdir -p /home/qun/data/flume/checkpoint

下载flume最新的包：

wgettar-zxvfapache-flume-1.6.0-bin.tar.gz

在slave3，slave4配置collectors

touch $FLUME_HOME/conf/server.conf

内容如下

a1.sources=r1a1.channels=c1a1.sinks=k1#setchannela1.channels.c1.type=filea1.channels.c1.checkpointDir=/home/qun/data/flume/checkpointa1.channels.c1.dataDirs=/home/qun/data/flume/data#othernode,nnatonnsa1.sources.r1.type=avroa1.sources.r1.bind=slave3a1.sources.r1.port=52020a1.sources.r1.interceptors=i1a1.sources.r1.interceptors.i1.type=statica1.sources.r1.interceptors.i1.key=Collectora1.sources.r1.interceptors.i1.value=SLAVE3a1.sources.r1.channels=c1#setsinktokafkaa1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.topic=mytopica1.sinks.k1.brokerList=kafkahost:9092a1.sinks.k1.requiredAcks=1a1.sinks.k1.batchSize=100a1.sinks.k1.channel=c1

在slave0，slave1，slave2配置agent

touch $FLUME_HOME/conf/client.conf

内容如下

agent1.channels=c1agent1.sources=r1agent1.sinks=k1k2#setgruopagent1.sinkgroups=g1#setchannelagent1.channels.c1.type=fileagent1.channels.c1.checkpointDir=/home/qun/data/flume/checkpointagent1.channels.c1.dataDirs=/home/qun/data/flume/dataagent1.sources.r1.channels=c1agent1.sources.r1.type=spooldiragent1.sources.r1.spoolDir=/home/qun/data/flume/logsagent1.sources.r1.fileHeader=falseagent1.sources.r1.interceptors=i1i2agent1.sources.r1.interceptors.i1.type=staticagent1.sources.r1.interceptors.i1.key=Typeagent1.sources.r1.interceptors.i1.value=LOGINagent1.sources.r1.interceptors.i2.type=timestamp#setsink1agent1.sinks.k1.channel=c1agent1.sinks.k1.type=avroagent1.sinks.k1.hostname=slave3agent1.sinks.k1.port=52020#setsink2agent1.sinks.k2.channel=c1agent1.sinks.k2.type=avroagent1.sinks.k2.hostname=slave4agent1.sinks.k2.port=52020#setsinkgroupagent1.sinkgroups.g1.sinks=k1k2#setfailoveragent1.sinkgroups.g1.processor.type=failoveragent1.sinkgroups.g1.processor.priority.k1=10agent1.sinkgroups.g1.processor.priority.k2=1agent1.sinkgroups.g1.processor.maxpenalty=10000

在slave3，slave4上启动collecters

flume-ngagent-na1-cconf-f/home/qun/apache-flume-1.6.0-bin/conf/server.conf-Dflume.root.logger=DEBUG,console

在slave0，slave1，slave2上启动agent

flume-ngagent-nagent1-cconf-f/home/qun/apache-flume-1.6.0-bin/conf/client.conf-Dflume.root.logger=DEBUG,console

测试功能

echo"helloflume">>/home/qun/data/flume/logs/test.txtcollectorslave3接收到agent的日志：16/05/2612:44:24INFOfile.EventQueueBackingStoreFile:Startcheckpointfor/home/qun/data/flume/checkpoint/checkpoint,elementstosync=216/05/2612:44:24INFOfile.EventQueueBackingStoreFile:Updatingcheckpointmetadata:logWriteOrderID:1464235734894,queueSize:0,queueHead:016/05/2612:44:24INFOfile.Log:Updatedcheckpointforfile:/home/qun/data/flume/data/log-3position:786logWriteOrderID:146423573489416/05/2612:44:24INFOfile.Log:Removingoldfile:/home/qun/data/flume/data/log-116/05/2612:44:24INFOfile.Log:Removingoldfile:/home/qun/data/flume/data/log-1.meta16/05/2612:44:54INFOfile.EventQueueBackingStoreFile:Startcheckpointfor/home/qun/data/flume/checkpoint/checkpoint,elementstosync=216/05/2612:44:54INFOfile.EventQueueBackingStoreFile:Updatingcheckpointmetadata:logWriteOrderID:1464235734901,queueSize:0,queueHead:016/05/2612:44:54INFOfile.Log:Updatedcheckpointforfile:/home/qun/data/flume/data/log-3position:1179logWriteOrderID:1464235734901

测试collecters Failover

杀死slave3的flume进程，kill -9 pid

echo"helloflume">>/home/qun/data/flume/logs/test.txtcollectorslave4接收到agent的日志：16/05/2612:08:27INFOfile.EventQueueBackingStoreFile:Startcheckpointfor/home/qun/data/flume/checkpoint/checkpoint,elementstosync=216/05/2612:08:27INFOfile.EventQueueBackingStoreFile:Updatingcheckpointmetadata:logWriteOrderID:1464234987484,queueSize:0,queueHead:016/05/2612:08:27INFOfile.Log:Updatedcheckpointforfile:/home/qun/data/flume/data/log-3position:393logWriteOrderID:146423498748416/05/2612:08:27INFOfile.LogFile:ClosingRandomReader/home/qun/data/flume/data/log-116/05/2612:54:38INFOclient.ClientUtils$:Fetchingmetadatafrombrokerid:0,host:xiaobin,port:9092withcorrelationid4for1topic(s)Set(mytopic)16/05/2612:54:38INFOproducer.SyncProducer:Connectedtoxiaobin:9092forproducing16/05/2612:54:38INFOproducer.SyncProducer:Disconnectingfromxiaobin:909216/05/2612:54:38INFOproducer.SyncProducer:Disconnectingfromxiaobin:909216/05/2612:54:38INFOproducer.SyncProducer:Connectedtoxiaobin:9092forproducing16/05/2612:54:57INFOfile.EventQueueBackingStoreFile:Startcheckpointfor/home/qun/data/flume/checkpoint/checkpoint,elementstosync=216/05/2612:54:57INFOfile.EventQueueBackingStoreFile:Updatingcheckpointmetadata:logWriteOrderID:1464234987491,queueSize:0,queueHead:016/05/2612:54:57INFOfile.Log:Updatedcheckpointforfile:/home/qun/data/flume/data/log-3position:786logWriteOrderID:146423498749116/05/2612:54:57INFOfile.Log:Removingoldfile:/home/qun/data/flume/data/log-116/05/2612:54:57INFOfile.Log:Removingoldfile:/home/qun/data/flume/data/log-1.meta

一会儿再写····