機(jī)器列表：

站在用戶的角度思考問題，與客戶深入溝通，找到陸港網(wǎng)站設(shè)計(jì)與陸港網(wǎng)站推廣的解決方案，憑借多年的經(jīng)驗(yàn)，讓設(shè)計(jì)與互聯(lián)網(wǎng)技術(shù)結(jié)合，創(chuàng)造個(gè)性化、用戶體驗(yàn)好的作品，建站類型包括：網(wǎng)站設(shè)計(jì)制作、網(wǎng)站設(shè)計(jì)、企業(yè)官網(wǎng)、英文網(wǎng)站、手機(jī)端網(wǎng)站、網(wǎng)站推廣、域名注冊(cè)、網(wǎng)站空間、企業(yè)郵箱。業(yè)務(wù)覆蓋陸港地區(qū)。

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

在每個(gè)機(jī)器上創(chuàng)建目錄

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

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

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

下載flume最新的包：

wget 
tar -zxvf apache-flume-1.6.0-bin.tar.gz

在slave3，slave4配置collectors

touch $FLUME_HOME/conf/server.conf

內(nèi)容如下

a1.sources = r1
a1.channels = c1
a1.sinks = k1
#set channel
a1.channels.c1.type = file
a1.channels.c1.checkpointDir=/home/qun/data/flume/checkpoint
a1.channels.c1.dataDirs=/home/qun/data/flume/data
# other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = slave3
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = SLAVE3
a1.sources.r1.channels = c1
#set sink to kafka
a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic=mytopic
a1.sinks.k1.brokerList=kafkahost:9092
a1.sinks.k1.requiredAcks=1
a1.sinks.k1.batchSize=100
a1.sinks.k1.channel=c1

在slave0，slave1，slave2配置agent

touch $FLUME_HOME/conf/client.conf

內(nèi)容如下

agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2
#set gruop
agent1.sinkgroups = g1 
#set channel
agent1.channels.c1.type = file
agent1.channels.c1.checkpointDir=/home/qun/data/flume/checkpoint
agent1.channels.c1.dataDirs=/home/qun/data/flume/data
agent1.sources.r1.channels = c1
agent1.sources.r1.type = spooldir
agent1.sources.r1.spoolDir=/home/qun/data/flume/logs
agent1.sources.r1.fileHeader = false
agent1.sources.r1.interceptors = i1 i2
agent1.sources.r1.interceptors.i1.type = static
agent1.sources.r1.interceptors.i1.key = Type
agent1.sources.r1.interceptors.i1.value = LOGIN
agent1.sources.r1.interceptors.i2.type = timestamp
# set sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = slave3
agent1.sinks.k1.port = 52020
# set sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = slave4
agent1.sinks.k2.port = 52020
#set sink group
agent1.sinkgroups.g1.sinks = k1 k2
#set failover
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000

在slave3，slave4上啟動(dòng)collecters

flume-ng agent -n a1 -c conf -f /home/qun/apache-flume-1.6.0-bin/conf/server.conf -Dflume.root.logger=DEBUG,console

在slave0，slave1，slave2上啟動(dòng)agent

flume-ng agent -n agent1 -c conf -f /home/qun/apache-flume-1.6.0-bin/conf/client.conf -Dflume.root.logger=DEBUG,console

測(cè)試功能

echo "hello flume">>/home/qun/data/flume/logs/test.txt

collector slave3 接收到agent的日志：

16/05/26 12:44:24 INFO file.EventQueueBackingStoreFile: Start checkpoint for /home/qun/data/flume/checkpoint/checkpoint, elements to sync = 2
16/05/26 12:44:24 INFO file.EventQueueBackingStoreFile: Updating checkpoint metadata: logWriteOrderID: 1464235734894, queueSize: 0, queueHead: 0
16/05/26 12:44:24 INFO file.Log: Updated checkpoint for file: /home/qun/data/flume/data/log-3 position: 786 logWriteOrderID: 1464235734894
16/05/26 12:44:24 INFO file.Log: Removing old file: /home/qun/data/flume/data/log-1
16/05/26 12:44:24 INFO file.Log: Removing old file: /home/qun/data/flume/data/log-1.meta
16/05/26 12:44:54 INFO file.EventQueueBackingStoreFile: Start checkpoint for /home/qun/data/flume/checkpoint/checkpoint, elements to sync = 2
16/05/26 12:44:54 INFO file.EventQueueBackingStoreFile: Updating checkpoint metadata: logWriteOrderID: 1464235734901, queueSize: 0, queueHead: 0
16/05/26 12:44:54 INFO file.Log: Updated checkpoint for file: /home/qun/data/flume/data/log-3 position: 1179 logWriteOrderID: 1464235734901

測(cè)試collecters Failover

殺死slave3的flume進(jìn)程，kill -9 pid

echo "hello flume">>/home/qun/data/flume/logs/test.txt
collector slave4 接收到agent的日志：
16/05/26 12:08:27 INFO file.EventQueueBackingStoreFile: Start checkpoint for /home/qun/data/flume/checkpoint/checkpoint, elements to sync = 2
16/05/26 12:08:27 INFO file.EventQueueBackingStoreFile: Updating checkpoint metadata: logWriteOrderID: 1464234987484, queueSize: 0, queueHead: 0
16/05/26 12:08:27 INFO file.Log: Updated checkpoint for file: /home/qun/data/flume/data/log-3 position: 393 logWriteOrderID: 1464234987484
16/05/26 12:08:27 INFO file.LogFile: Closing RandomReader /home/qun/data/flume/data/log-1
16/05/26 12:54:38 INFO client.ClientUtils$: Fetching metadata from broker id:0,host:xiaobin,port:9092 with correlation id 4 for 1 topic(s) Set(mytopic)
16/05/26 12:54:38 INFO producer.SyncProducer: Connected to xiaobin:9092 for producing
16/05/26 12:54:38 INFO producer.SyncProducer: Disconnecting from xiaobin:9092
16/05/26 12:54:38 INFO producer.SyncProducer: Disconnecting from xiaobin:9092
16/05/26 12:54:38 INFO producer.SyncProducer: Connected to xiaobin:9092 for producing
16/05/26 12:54:57 INFO file.EventQueueBackingStoreFile: Start checkpoint for /home/qun/data/flume/checkpoint/checkpoint, elements to sync = 2
16/05/26 12:54:57 INFO file.EventQueueBackingStoreFile: Updating checkpoint metadata: logWriteOrderID: 1464234987491, queueSize: 0, queueHead: 0
16/05/26 12:54:57 INFO file.Log: Updated checkpoint for file: /home/qun/data/flume/data/log-3 position: 786 logWriteOrderID: 1464234987491
16/05/26 12:54:57 INFO file.Log: Removing old file: /home/qun/data/flume/data/log-1
16/05/26 12:54:57 INFO file.Log: Removing old file: /home/qun/data/flume/data/log-1.meta

一會(huì)兒再寫····

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場(chǎng)景需求。

本文標(biāo)題：flume-1.6.0高可用測(cè)試&&數(shù)據(jù)入Kafka-創(chuàng)新互聯(lián)
文章源于：http://www.weahome.cn/article/gjoed.html