flume-1.6.0高可用性テスト&&データ入Kafka

5897 ワード
テスト 高可用性 flume
マシンのリスト:
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）

各マシンにディレクトリを作成
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

slave 3では、slave 4はcollectorsを構成します.
touch $FLUME_HOME/conf/server.conf
内容は以下の通り
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

 
slave 0,slave 1,slave 2でagentを構成する
touch $FLUME_HOME/conf/client.conf
内容は以下の通り
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

slave 3、slave 4で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

slave 0,slave 1,slave 2で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

テスト機能
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

collecters Failoverのテスト
slave 3を殺すflumeプロセス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

あとで書きます・・・
BOJ 11050異項係数[Java]
Git flow