ambari下的flume和kafka整合

 1 #扫描指定文件配置
 2 agent.sources = s1    
 3 agent.channels = c1  
 4 agent.sinks = k1
 5 
 6 agent.sources.s1.type=exec
 7 agent.sources.s1.command=tail -F /home/flume/test/test.log
 8 agent.sources.s1.channels=c1
 9 agent.channels.c1.type=memory
10 agent.channels.c1.capacity=10000
11 agent.channels.c1.transactionCapacity=100
12 
13 #设置Kafka接收器
14 agent.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
15 #设置Kafka的broker地址和端口号
16 agent.sinks.k1.brokerList=172.16.38.159:6667
17 #设置Kafka的Topic
18 agent.sinks.k1.topic=test
19 #设置序列化方式
20 agent.sinks.k1.serializer.class=kafka.serializer.StringEncoder
21 
22 agent.sinks.k1.channel=c1

#创建一个test的topic
bin/kafka-topics.sh --zookeeper 172.16.38.159:2181 --create --topic test --replication-factor 1 --partitions 1

java -cp KafkaOffsetMonitor-assembly-0.2.0.jar  com.quantifind.kafka.offsetapp.OffsetGetterWeb  --zk 172.16.38.159:2181  --port 8089  --refresh 10.seconds  --retain 1.days

./kafka-console-consumer.sh -zookeeper 172.16.38.159:2181 --from-beginning --topic test