flume kafka channel 테스트

kafka를 flume channel로 이용해 보자.

 

 

 

먼저 channel을 만든다. 순서보장을 위해 parition은 1개만 선언하고, ha용 replication factor는 2로 실행

[kafka@big-slave4 ~]$ kafka-topics.sh \
--zookeeper big-master:2181,big-slave1:2181,big-slave2:2181/kafka-cluster \
--topic flume-channel --partitions 1 --replication-factor 2 --create
Created topic "flume-channel".

 

leader 확인

[kafka@big-slave4 ~]$ kafka-topics.sh \
--zookeeper big-master:2181,big-slave1:2181,big-slave2:2181/kafka-cluster \
--topic flume-channel --describe
Topic:flume-channel     PartitionCount:1        ReplicationFactor:2     Configs:
        Topic: flume-channel    Partition: 0    Leader: 1       Replicas: 1,3   Isr: 1,3

 

 

[kafka@big-slave4 ~]$ kafka-console-consumer.sh  \
--bootstrap-server big-slave2:9092,big-slave3:9092,big-slave4:9092 \
--topic flume-channel --from-beginning

^CProcessed a total of 0 messages
[kafka@big-slave4 ~]$ kafka-consumer-groups.sh  \
--bootstrap-server big-slave2:9092,big-slave3:9092,big-slave4:9092 \
--list
Note: This will not show information about old Zookeeper-based consumers.
KMOffsetCache-big-master
console-consumer-94532
[kafka@big-slave4 ~]$ kafka-consumer-groups.sh  \
--bootstrap-server big-slave2:9092,big-slave3:9092,big-slave4:9092 \
--group console-consumer-94532 --delete
Note: This will not show information about old Zookeeper-based consumers.
Deletion of requested consumer groups ('console-consumer-94532') was successful.

[kafka@big-slave4 ~]$ kafka-console-consumer.sh  \
--bootstrap-server big-slave2:9092,big-slave3:9092,big-slave4:9092 \
--topic flume-channel \
--group flume-channel-consumers \
--from-beginning

 

client flume conf

#Source
agent.sources = randomGen
agent.sources.randomGen.type = com.dbility.bigdata.flume.source.RandomDataGenerator
agent.sources.randomGen.batchSize = 10
agent.sources.randomGen.channels = kafkaChannel

#Channel
agent.channels = kafkaChannel
agent.channels.kafkaChannel.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafkaChannel.kafka.bootstrap.servers = big-slave2:9092,big-slave3:9092,big-slave4:9092
agent.channels.kafkaChannel.kafka.topic = flume-channel
agent.channels.kafkaChannel.kafka.consumer.group.id = flume-channe1-consumers

client agent 실행

E:\apache-flume-1.8.0-bin>bin\flume-ng agent -n agent -c conf -f conf\randomGen2.properties

E:\apache-flume-1.8.0-bin>set JAVA_HOME=E:\apache-flume-1.8.0-bin\jdk180\jre

E:\apache-flume-1.8.0-bin>set JAVA_OPTS=" -Xmx256M"

E:\apache-flume-1.8.0-bin>powershell.exe -NoProfile -InputFormat none -ExecutionPolicy unrestricted -File E:\apache-flume-1.8.0-bin\bin\flume-ng.ps1 agent -n agent -c conf -f conf\randomGen2.properties

WARN: Config directory not set. Defaulting to E:\apache-flume-1.8.0-bin\conf
Sourcing environment configuration script E:\apache-flume-1.8.0-bin\conf\flume-env.ps1
WARN: Did not find E:\apache-flume-1.8.0-bin\conf\flume-env.ps1
WARN: HADOOP_PREFIX or HADOOP_HOME not found
WARN: HADOOP_PREFIX not set. Unable to include Hadoop's classpath & java.library.path
WARN: HBASE_HOME not found
WARN: HIVE_HOME not found

  Running FLUME agent :
    class: org.apache.flume.node.Application
    arguments: -n agent -f "E:\apache-flume-1.8.0-bin\conf\randomGen2.properties"