flume的自定義sink-Kafka

    1、創建一個agent，sink類型需指定為自定義sink
        vi /usr/local/flume/conf/agent3.conf
        agent3.sources=as1
        agent3.channels=c1
        agent3.sinks=s1

        agent3.sources.as1.type=avro
        agent3.sources.as1.bind=0.0.0.0
        agent3.sources.as1.port=41414
        agent3.sources.as1.channels=c1

        agent3.channels.c1.type=memory

        agent3.sinks.s1.type=storm.test.kafka.TestKafkaSink
        agent3.sinks.s1.channel=c1
    2、創建自定義kafka sink（自定義kafka sink中包裝的是kafka的生產者），代碼如下
        //參考flume官方的開發文檔：http://flume.apache.org/FlumeDeveloperGuide.html#sink
        //自定義kafkasink需要繼承AbstractSink類實現Configurable接口
        //該sink中使用的kafka topic（test111）必須存在      

package storm.test.kafka;

        import java.util.Properties;

        import kafka.javaapi.producer.Producer;
        import kafka.producer.KeyedMessage;
        import kafka.producer.ProducerConfig;
        import kafka.serializer.StringEncoder;

        import org.apache.flume.Channel;
        import org.apache.flume.Context;
        import org.apache.flume.Event;
        import org.apache.flume.EventDeliveryException;
        import org.apache.flume.Transaction;
        import org.apache.flume.conf.Configurable;
        import org.apache.flume.sink.AbstractSink;

        public class TestKafkaSink extends AbstractSink implements Configurable {

            Producer<String, String> producer;
            String topic = "test111";
            
            @Override
            public Status process() throws EventDeliveryException {
                Status status = null;
                Channel channel = getChannel();
                Transaction transaction = channel.getTransaction();
                transaction.begin();
                try {
                    Event event = channel.take();
                    if (event==null) {
                        transaction.rollback();
                        status = Status.BACKOFF;
                        return status;
                    }
                    byte[] body = event.getBody();
                    final String msg = new String(body);
                    final KeyedMessage<String, String> message = new KeyedMessage<String, String>(topic , msg);
                    producer.send(message);
                    transaction.commit();
                    status = Status.READY;            
                } catch (Exception e) {
                    transaction.rollback();
                    status = Status.BACKOFF;
                } finally {
                    transaction.close();
                }
                
                return status;
            }

            @Override
            public void configure(Context arg0) {
                Properties prop = new Properties();
                prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
                prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
                prop.put("serializer.class", StringEncoder.class.getName());
                producer = new Producer<String, String>(new ProducerConfig(prop));
            }

        }

        將代碼打包為kafkasink.jar后復制到flume所在節點上的flume/lib目錄下，然后還需要將kafka_2.10-0.8.2.0.jar、kafka-clients-0.8.2.0.jar、metrics-core-2.2.0.jar、scala-library-2.10.4.jar這4個jar包復制到flume所在節點上的flume/lib目錄下。
    3、啟動flume自定義的kafkasink的agent
        [root@h5 ~]# cd /usr/local/flume/
        [root@h5 flume]# bin/flume-ng agent --conf conf/ --conf-file conf/agent3.conf --name agent3 -Dflume.root.logger=INFO,console
    4、將日志寫入到flume的agent，代碼如下
        log4j.properties
            log4j.rootLogger=INFO,flume
            log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
            log4j.appender.flume.Hostname = 192.168.1.35
            log4j.appender.flume.Port = 41414
            log4j.appender.flume.UnsafeMode = true
        將日志寫入到flume，代碼如下          

package com.mengyao.flume;

            import java.io.File;
            import java.io.IOException;
            import java.util.Collection;
            import java.util.List;

            import org.apache.commons.io.FileUtils;
            import org.apache.log4j.Logger;

            public class FlumeProducer {

                private static List<String> getLines() {
                    List<String> lines = null;
                    try {
                        final Collection<File> listFiles = FileUtils.listFiles(new File("D:/"), null, false);
                        for (File file : listFiles) {
                            lines = FileUtils.readLines(file);
                            break;
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                    return lines;
                }
                
                public static void main(String[] args) throws Exception {
                    final List<String> lines = getLines();
                    final Logger logger = Logger.getLogger(FlumeProducer.class);
                    for (String line : lines) {
                        logger.info(line+"\t"+System.currentTimeMillis());
                        Thread.sleep(1000);            
                    }
                }    
            }

            必須加入flume-ng-log4jappender-1.5.0-cdh5.1.3-jar-with-dependencies.jar這個依賴jar
    5、使用kafka消費者消費flume（自定義kafka sink中使用了kafka的生產者）生產的數據
        1、消費者shell代碼
            [root@h7 kafka]# bin/kafka-console-consumer.sh --zookeeper h7:2181 --topic test111 --from-beginning        ##kafka集群是h5、h6、h7;zookeeper集群是h5、h6、h7。在任意kafka節點上使用消費者都一樣
        
        2、消費者java代碼        

package storm.test.kafka;

            import java.util.HashMap;
            import java.util.List;
            import java.util.Map;
            import java.util.Properties;

            import kafka.consumer.Consumer;
            import kafka.consumer.ConsumerConfig;
            import kafka.consumer.ConsumerIterator;
            import kafka.consumer.KafkaStream;
            import kafka.javaapi.consumer.ConsumerConnector;
            import kafka.serializer.StringEncoder;

            public class TestConsumer {

                static final String topic = "test111";
                
                public static void main(String[] args) {
                    Properties prop = new Properties();
                    prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
                    prop.put("serializer.class", StringEncoder.class.getName());
                    prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
                    prop.put("group.id", "group1");
                    ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
                    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
                    topicCountMap.put(topic, 1);
                    Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
                    final KafkaStream<byte[], byte[]> kafkaStream = messageStreams.get(topic).get(0);
                    ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
                    while (iterator.hasNext()) {
                        String msg = new String(iterator.next().message());
                        System.out.println("收到消息："+msg);
                    }
                }

            }