案例编码实现

package com.atguigu;

import org.apache.flume.*;

import org.apache.flume.conf.Configurable;

import org.apache.flume.sink.AbstractSink;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

public class MySink extends AbstractSink implements Configurable {

    //创建Logger对象

    private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class);

    private String prefix;

    private String suffix;

    @Override

    public Status process() throws EventDeliveryException {

        //声明返回值状态信息

        Status status;

        //获取当前Sink绑定的Channel

        Channel ch = getChannel();

        //获取事务

        Transaction txn = ch.getTransaction();

        //声明事件

        Event event;

        //开启事务

        txn.begin();

        //读取Channel中的事件，直到读取到事件结束循环

        while (true) {

            event = ch.take();

            if (event != null) {

                break;

            }

        }

        try {

            //处理事件（打印）

            LOG.info(prefix + new String(event.getBody()) + suffix);

            //事务提交

            txn.commit();

            status = Status.READY;

        } catch (Exception e) {

            //遇到异常，事务回滚

            txn.rollback();

            status = Status.BACKOFF;

        } finally {

            //关闭事务

            txn.close();

        }

        return status;

    }

    @Override

    public void configure(Context context) {

        //读取配置文件内容，有默认值

        prefix = context.getString(“prefix”, “hello:”);

        //读取配置文件内容，无默认值

        suffix = context.getString(“suffix”);

    }

}

想要了解跟多关于大数据培训课程内容欢迎关注尚硅谷大数据，尚硅谷除了这些技术文章外还有免费的高质量大数据培训课程视频供广大学员下载学习

上一篇: 大数据培训技术自定义Sink
下一篇: 前端培训技术AngularJS 指令