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));
    	}
    }
    

      

  • 相关阅读:
    【Python3 爬虫】U31_selenium定位元素
    【Python3 爬虫】U30_selenium初识
    verilog语法实例学习(1)
    verilog语法学习目录
    在testbench从文件读入激励
    Verilog 加法器和减法器(7)
    浮点数的表示和运算
    Verilog 加法器和减法器(6)
    Verilog 加法器和减法器(5)
    Verilog 加法器和减法器(4)
  • 原文地址:https://www.cnblogs.com/zqzdong/p/6438958.html
Copyright © 2011-2022 走看看