Flume 自定义组件

public class CustomSource extends AbstractSource implements Configurable, PollableSource{
    private String prefix;
    private String suffix;

    public void configure(Context context) {
        prefix = context.getString("prefix","danner");
        suffix = context.getString("suffix");
    }
    public Status process() throws EventDeliveryException {
        Status status = null;

        try {
            // 创建 Event
            SimpleEvent event = new SimpleEvent();

            // event 添加内容
            event.setBody(new String(prefix + new Random().nextInt(100) + suffix).getBytes());

            // event 向后传递到 Channel
            getChannelProcessor().processEvent(event);

            // 处理成功
            status = Status.READY;
        }catch (Throwable t){
            t.printStackTrace();
            status = Status.BACKOFF;
        }

        // 防止太快
        try {
            Thread.sleep(2*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return status;
    }
    ......

public class CustomSink extends AbstractSink implements Configurable {
    private static final Logger logger = LoggerFactory
            .getLogger(CustomSink.class);
    private String prefix;
    private String suffix;

    public void configure(Context context) {
        prefix = context.getString("prefix","hz");
        suffix = context.getString("suffix","");
    }
    public Status process() throws EventDeliveryException {

        Status status = null;
        Event event = null;
        // 得到输入的 channel
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        try {
            transaction.begin();
            // 获取 event
            while (true){
                event = channel.take();
                if (event != null){
                    break;
                }
            }
            // 输出 event
            logger.info("custom sink: " + new String(event.getBody()));
			// 事务提交
            transaction.commit();
            status = Status.READY;
        }catch (Throwable t){
            t.printStackTrace();
            // event 处理失败，回滚
            transaction.rollback();
            status = Status.BACKOFF;
        }finally {
            // 关闭
            transaction.close();
        }
        return status;
    }
 }

private LinkedBlockingDeque<Event> takeList; // sink 从 channel 取出 event 的记录，回滚使用
private LinkedBlockingDeque<Event> putList;	 // source 放入 channel 的 event

public class CustomInterceptor implements Interceptor {
    private List<Event> list;
    public void initialize() {
        list = new LinkedList<Event>();
    }

    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();
        String content = new String(event.getBody());
        // selector 需要配置 type  header
        if(content.contains("bejson")){
            headers.put("type","bejson");
        }else{
            headers.put("type","other");
        }
        return event;
    }

    public List<Event> intercept(List<Event> events) {
        list.clear();
        for(Event event :events){
            list.add(intercept(event));
        }
        return list;
    }

    public static class Builder implements Interceptor.Builder{
        public Interceptor build() {
            return new CustomInterceptor();
        }
        public void configure(Context context) {

        }
    }
    public void close() {
    }
}