Flume 基础学习

package flume_interceptor;

import org.apache.flume.Context;

import org.apache.flume.Event;

import org.apache.flume.interceptor.Interceptor;

import java.util.List;

import java.util.Map;

/**

 * @author lianzhipeng

 * @Description

 * @create 2020-05-05 10:45

 */

public class MyInterceptor implements Interceptor {

    /**

     * Description: 初始化方法，新建 Interceptor 时使用

     *

     * @Author: lianzhipeng

     * @Date: 2020/5/5 10:45

     * @return: void

     */

    public void initialize() {

    }

    /**

     * Description: 更改方法，对 event 进行处理

     *

     * @param event 传入的数据

     * @Author: lianzhipeng

     * @Date: 2020/5/5 10:47

     * @return: org.apache.flume.Event 返回处理好的数据

     */

    public Event intercept(Event event) {

        // 获取 event 的 header

        Map<String, String> headers = event.getHeaders();

        // 获取 event 的 body

        byte[] body = event.getBody();

        // 处理数据

        String string = new String(body);

        char c = string.charAt(0);

        if (c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z') {

            headers.put("type", "char");

        } else {

            headers.put("type", "not-char");

        }

        // 返回数据

        return event;

    }

    public List<Event> intercept(List<Event> list) {

        for (Event event : list) {

            intercept(event);

        }

        return list;

    }

    public void close() {

    }

    /**

     * 框架会调用 MyBulider 来创建自定义拦截器实例

     */

    public static class MyBulider implements Builder {

        /**

         * Description: 创建自定义拦截器实例的方法

         *

         * @Author: lianzhipeng

         * @Date: 2020/5/5 10:54

         * @return: org.apache.flume.interceptor.Interceptor

         */

        public Interceptor build() {

            return new MyInterceptor();

        }

        /**

         * Description: 读取配置信息

         *

         * @param context

         * @Author: lianzhipeng

         * @Date: 2020/5/5 10:54

         * @return: void

         */

        public void configure(Context context) {

        }

    }

}

package flume_interceptor;

import org.apache.flume.Context;

import org.apache.flume.Event;

import org.apache.flume.EventDeliveryException;

import org.apache.flume.PollableSource;

import org.apache.flume.channel.ChannelProcessor;

import org.apache.flume.conf.Configurable;

import org.apache.flume.event.SimpleEvent;

import org.apache.flume.source.AbstractSource;

/**

 * @author lianzhipeng

 * @Description

 * @create 2020-05-05 14:31

 */

public class MySource extends AbstractSource implements Configurable, PollableSource {

    private String prefix;

    private Long interval;

    /**

     * Description: 拉取事件并交给 ChannelProcessor 处理的方法

     *

     * @Author: lianzhipeng

     * @Date: 2020/5/5 14:33

     * @return: org.apache.flume.PollableSource.Status

     */

    public Status process() throws EventDeliveryException {

        Status status = null;

        try {

            // 通过外部方法拉取数据

            Event e = getSomeData();

            // Store the Event into this Source's associated Channel(s)

            ChannelProcessor channelProcessor = getChannelProcessor();

            channelProcessor.processEvent(e);

            status = Status.READY;

        } catch (Throwable t) {

            // Log exception, handle individual exceptions as needed

            status = Status.BACKOFF;

            // re-throw all Errors

            if (t instanceof Error) {

                throw (Error)t;

            }

        }

        return status;

    }

    /**

     * Description: 拉取数据并包装成 event 的过程

     * @Author: lianzhipeng

     * @Date: 2020/5/5 14:55

     * @return: org.apache.flume.Event 拉取到的数据

    */

    private Event getSomeData() throws InterruptedException {

        int i = (int) (Math.random() * 1000);

        // 添加前缀

        String message = prefix + i ;

        Thread.sleep(interval);

        //

        SimpleEvent event = new SimpleEvent();

        event.setBody(message.getBytes());

        return  event;

    }

    /**

     * Description: 如果拉取不到数据，backoff 时间的增长速度

     *

     * @Author: lianzhipeng

     * @Date: 2020/5/5 14:34

     * @return: long 增长量

     */

    public long getBackOffSleepIncrement() {

        return 1000;

    }

    /**

     * Description: 最大的等待时间

     *

     * @Author: lianzhipeng

     * @Date: 2020/5/5 14:38

     * @return: long

     */

    public long getMaxBackOffSleepInterval() {

        return 10000;

    }

    /**

     * Description: 配置参数，来自于 configurable，可以定义我们自己定义的 source

     *

     * @param context 配置文件

     * @Author: lianzhipeng

     * @Date: 2020/5/5 14:39

     * @return: void

     */

    public void configure(Context context) {

        prefix = context.getString("prefff","xxxx" );

        interval = context.getLong("interval",500L);

    }

}

package flume_interceptor;

import org.apache.flume.*;

import org.apache.flume.conf.Configurable;

import org.apache.flume.sink.AbstractSink;

import java.io.IOException;

/**

 * @author lianzhipeng

 * @Description

 * @create 2020-05-05 14:31

 */

public class MySiink extends AbstractSink implements Configurable {

    /**

     * Description: 改方法调用时，会从 Channel 中拉取数据并处理

     *

     * @Author: lianzhipeng

     * @Date: 2020/5/5 15:09

     * @return: org.apache.flume.Sink.Status 处理的状态

     */

    public Status process() throws EventDeliveryException {

        Status status = null;

        // Start transaction

        // 获取 channel

        Channel ch = getChannel();

        // 拉取数据的事务

        Transaction txn = ch.getTransaction();

        // 开始拉取

        txn.begin();

        try {

            // This try clause includes whatever Channel operations you want to do

            // 拉取的数据，如果拉取不到，则返回 null

            Event event;

            // 如果拉取的数据为 null，则等 0.1 秒后继续拉取数据，知道拉取数据

            while ((event = ch.take()) == null) {

                Thread.sleep(100);

            }

            // Send the Event to the external repository.

            // 如果拉取到了数据，将数据进行处理

            storeSomeData(event);

            txn.commit();

            status = Status.READY;

        } catch (Throwable t) {

            txn.rollback();

            // Log exception, handle individual exceptions as needed

            status = Status.BACKOFF;

            // re-throw all Errors

            if (t instanceof Error) {

                throw (Error) t;

            }

        } finally {

            // 拉取事务的关闭

            txn.close();

        }

        return status;

    }

    private void storeSomeData(Event event) throws IOException {

        // 获取 event 的 body 数据

        byte[] body = event.getBody();

        // 将数据写出到控制台

        System.out.write(body);

        System.out.println();

    }

    public void configure(Context context) {

    }

}