Flume五：自定义source

刺骨的言语ヽ痛彻心扉 2023-06-30 13:56 6阅读 0赞

# 案例 #

> ### 自定义source实现类 ###
> 
>     package com.flume.source;
>     
>     import org.apache.flume.Context;
>     import org.apache.flume.EventDeliveryException;
>     import org.apache.flume.PollableSource;
>     import org.apache.flume.conf.Configurable;
>     import org.apache.flume.event.SimpleEvent;
>     import org.apache.flume.source.AbstractSource;
>     
>     /**
>      * Created by HP on 2020/1/17.
>      */
>     public class MySource extends AbstractSource implements Configurable,PollableSource{
>         //定义全局的前缀&后缀
>         private String prefix;
>         private String subfix;
>         private int delay;
>     
>     
>         /**
>          * 1、接受数据(自定义业务数据来源，可为JDBC或者本地/网络文件系统等等等)
>          * 2、封装为事件
>          * 3、将事件传给channel
>          * @return
>          * @throws EventDeliveryException
>          */
>         @Override
>         public Status process() throws EventDeliveryException {
>             Status status=null;
>             try {
>                 //1、接受数据 模拟创建100条数据，每条数据的生成睡眠2秒中
>                 for (int i = 0; i < 100; i++) {
>                     //2、构建事件对象
>                     SimpleEvent event=new SimpleEvent();
>                     //3、给事件设置值
>                     event.setBody((prefix+"--"+i+"--"+subfix).getBytes());
>                     //4、将事件传给channel
>                     getChannelProcessor().processEvent(event);
>                     //设置每次循环的间隔事件，防止数据写入太快，不方便测试
>                     Thread.sleep(delay);
>     
>                     status=status.READY;//就绪状态
>                 }
>             } catch (InterruptedException e) {
>                 e.printStackTrace();
>                 status=status.BACKOFF;//退避状态
>             }
>             return status;
>         }
>     
>         @Override
>         public long getBackOffSleepIncrement() {
>             return 0;
>         }
>     
>         @Override
>         public long getMaxBackOffSleepInterval() {
>             return 0;
>         }
>     
>         @Override
>         public void configure(Context context) {
>             //读取配置信息，给前后缀赋值
>             prefix=context.getString("prefix");
>             subfix=context.getString("subfix","+++++");
>             delay=context.getInteger("delay",1000);
>         }
>     }
> 
> ### 打包上传到/opt/apache-flume-1.7.0-bin/ ###
> 
> ## agent配置文件 ##
> 
>     #定义
>     a1.sources = r1
>     a1.sinks = k1
>     a1.channels = c1
>     
>     #source
>     a1.sources.r1.type = com.flume.source.MySource
>     a1.sources.r1.prefix = HAHA
>     a1.sources.r1.subfix = HEHE
>     a1.sources.r1.delay = 1500
>     
>     #channel
>     a1.channels.c1.type = memory
>     a1.channels.c1.capacity = 1000
>     a1.channels.c1.transactionCapacity = 100
>     
>     #sink
>     #控制台日志打印
>     a1.sinks.k1.type = logger
>     
>     #关联关系
>     a1.sources.r1.channels = c1
>     a1.sinks.k1.channel = c1