Kafka与FlumeNG整合

1，作为Producer的Flume端配置，其中是以netcat为source数据源，sink是kafka

[html] 
    view plain
    copy
     
    
 
    
 hadoop@stormspark:~/bigdata/apache-flume-1.4.0-bin$ cat conf/producer1.properties   
 #agent section  
 producer.sources = s  
 producer.channels = c  
 producer.sinks = r  
   
 #source section  
 #producer.sources.s.type = seq  
 producer.sources.s.type = netcat  
 producer.sources.s.bind = localhost  
 producer.sources.s.port = 44444  
 producer.sources.s.channels = c  
   
 # Each sink's type must be defined  
 producer.sinks.r.type = org.apache.flume.plugins.KafkaSink  
 producer.sinks.r.metadata.broker.list=127.0.0.1:9092  
 producer.sinks.r.partition.key=0  
 producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition  
 producer.sinks.r.serializer.class=kafka.serializer.StringEncoder  
 producer.sinks.r.request.required.acks=0  
 producer.sinks.r.max.message.size=1000000  
 producer.sinks.r.producer.type=sync  
 producer.sinks.r.custom.encoding=UTF-8  
 producer.sinks.r.custom.topic.name=test  
   
 #Specify the channel the sink should use  
 producer.sinks.r.channel = c  
   
 # Each channel's type is defined.  
 producer.channels.c.type = memory  
 producer.channels.c.capacity = 1000  
   
 hadoop@stormspark:~/bigdata/apache-flume-1.4.0-bin$   

2，配置consumer，source是Kafka，sink是logger

[html] 
    view plain
    copy
     
    
 
    
 hadoop@stormspark:~/bigdata/apache-flume-1.4.0-bin$ cat conf/comsumer1.properties   
 umer config  
 ###########################################  
   
 consumer.sources = s  
 consumer.channels = c  
 consumer.sinks = r  
   
 consumer.sources.s.type = seq  
 consumer.sources.s.channels = c  
 consumer.sinks.r.type = logger  
   
 consumer.sinks.r.channel = c  
 consumer.channels.c.type = memory  
 consumer.channels.c.capacity = 100  
   
 consumer.sources.s.type = org.apache.flume.plugins.KafkaSource  
 consumer.sources.s.zookeeper.connect=127.0.0.1:2181  
 consumer.sources.s.group.id=testGroup  
 consumer.sources.s.zookeeper.session.timeout.ms=400  
 consumer.sources.s.zookeeper.sync.time.ms=200  
 consumer.sources.s.auto.commit.interval.ms=1000  
 consumer.sources.s.custom.topic.name=test  
 consumer.sources.s.custom.thread.per.consumer=4  

3,分别运行着两个agent

[html] 
    view plain
    copy
     
 bin/flume-ng agent --conf conf  --conf-file conf/producer1.properties --name producer -Dflume.root.logger=INFO,console

[html] 
    view plain
    copy
     
 bin/flume-ng agent --conf conf  --conf-file conf/comsumer1.properties   --name consumer -Dflume.root.logger=INFO,console

4，这时telnet上端口44444

[html] 
    view plain
    copy
     
    
 
    
 hadoop@stormspark:~/bigdata/apache-flume-1.4.0-bin$ telnet localhost 44444  
 Trying ::1...  
 Trying 127.0.0.1...  
 Connected to localhost.  
 Escape character is '^]'.  
 1111111111111111  
 OK  
 kak^Hfkakakkakakakkakkakkaakaknnnm  
 OK  
 abcdefghijklmnopqrstuvwxyz  
 OK  

两个agent都有信息输出

[html] 
    view plain
    copy
     
    
 
    
 2014-01-15 20:01:05,047 (lifecycleSupervisor-1-1) [INFO - kafka.utils.Logging$class.info(Logging.scala:67)] Property metadata.broker.list is overridden to 127.0.0.1:9092  
 ] -- [{ headers:{} body: 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 1111111111111111 }]lume.plugins.KafkaSink.process(KafkaSink.java:137)] Send Message to Kafka : [1111111111111111  
 2014-01-15 20:01:35,702 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:67)] Fetching metadata from broker id:0,host:127.0.0.1,port:9092 with correlation id 0 for 1 topic(s) Set(test)  
 2014-01-15 20:01:35,704 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:67)] Connected to 127.0.0.1:9092 for producing  
 2014-01-15 20:01:35,727 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:67)] Disconnecting from 127.0.0.1:9092  
 2014-01-15 20:01:35,767 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - kafka.utils.Logging$class.info(Logging.scala:67)] Connected to stormspark:9092 for producing  
 ] -- [{ headers:{} body: 6B 61 6B 08 66 6B 61 6B 61 6B 6B 61 6B 61 6B 61 kak.fkakakkakaka }]lume.plugins.KafkaSink.process(KafkaSink.java:137)] Send Message to Kafka : [kafkakakkakakakkakkakkaakaknnnm  
 ] -- [{ headers:{} body: 61 62 63 64 65 66 67 68 69 6A 6B 6C 6D 6E 6F 70 abcdefghijklmnop }]lume.plugins.KafkaSink.process(KafkaSink.java:137)] Send Message to Kafka : [abcdefghijklmnopqrstuvwxyz  
 ] -- [{ headers:{} body: 71 75 69 74 0D                                  quit. }]rg.apache.flume.plugins.KafkaSink.process(KafkaSink.java:137)] Send Message to Kafka : [quit  

另一个

[html] 
    view plain
    copy
     
    
 
    
 2014-01-15 19:58:02,434 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 2222222222222222 }  
 ]014-01-15 20:01:35,771 (pool-4-thread-1) [INFO - org.apache.flume.plugins.KafkaSource$ConsumerWorker.run(KafkaSource.java:230)] Receive Message [Thread 0: 1111111111111111  
 2014-01-15 20:01:36,487 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 1111111111111111 }  
 ]014-01-15 20:02:13,784 (pool-4-thread-1) [INFO - org.apache.flume.plugins.KafkaSource$ConsumerWorker.run(KafkaSource.java:230)] Receive Message [Thread 0: kafkakakkakakakkakkakkaakaknnnm  
 2014-01-15 20:02:14,500 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 6B 61 6B 08 66 6B 61 6B 61 6B 6B 61 6B 61 6B 61 kak.fkakakkakaka }  
 ]014-01-15 20:02:28,960 (pool-4-thread-1) [INFO - org.apache.flume.plugins.KafkaSource$ConsumerWorker.run(KafkaSource.java:230)] Receive Message [Thread 0: abcdefghijklmnopqrstuvwxyz  
 2014-01-15 20:02:29,506 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 61 62 63 64 65 66 67 68 69 6A 6B 6C 6D 6E 6F 70 abcdefghijklmnop }  
 ]014-01-15 20:03:54,986 (pool-4-thread-1) [INFO - org.apache.flume.plugins.KafkaSource$ConsumerWorker.run(KafkaSource.java:230)] Receive Message [Thread 0: quit  
 2014-01-15 20:03:55,529 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 71 75 69 74 0D                                  quit. }  
 ^C2014-01-15 20:09:10,094 (agent-shutdown-hook) [INFO - org.apache.flume.lifecycle.LifecycleSup  