zoukankan      html  css  js  c++  java
  • 自定义flume的sink

    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 FlumeToKafkaSink extends AbstractSink implements Configurable {
    	Producer<String, String> producer;
    	String topic = "zz";
    	@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);
    			String str[] = msg.split("#XT#");
    			if(str[str.length - 1].equals("td")){
    				final KeyedMessage<String, String> message = new KeyedMessage<String, String>(topic , msg);
    				producer.send(message);
    				transaction.commit();
    				status = Status.READY;  
    			}else{
    				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", "Master:2181,Slave:2181,Slave2:2181");
    		prop.put("metadata.broker.list", "Master:9092,Slave:9092,Slave2:9092");
    		prop.put("serializer.class", StringEncoder.class.getName());
    		producer = new Producer<String, String>(new ProducerConfig(prop));
    	}
    }
    

      

  • 相关阅读:
    C++ primer学习方法
    windows 下安装使用ipython
    读书和思考
    win7 64位 python3.4&opencv3.0配置安装
    Deep Residual Learning for Image Recognition(MSRA-深度残差学习)
    MATLAB 常用形态学操作函数
    形态学图像处理
    限制对比度自适应直方图均衡(Contrast Limited Adaptive histgram equalization/CLAHE)
    计算机视觉,机器学习 ( 一些资源)
    对CNN模块的分析
  • 原文地址:https://www.cnblogs.com/zqzdong/p/6438958.html
Copyright © 2011-2022 走看看