Flume经典案例之自定义Source

使用flume接收数据，并给每条数据添加前缀，输出到控制台。

1、打开IDEA，新建maven工程。

2、导入依赖

      org.apache.flume
      flume-ng-core
      1.6.0
    

3、编写代码

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;

import java.util.ArrayList;
import java.util.List;

public class MySource extends AbstractSource implements Configurable, PollableSource {

    private String prefix;
    
    @Override
    public Status process() throws EventDeliveryException {
        List list = new ArrayList<>();
        Status status = Status.READY;
        for (int i = 0; i < 5; i++) {
            //创建event的对象
            SimpleEvent event = new SimpleEvent();
            //给event中设置数据+前缀
            event.setBody((prefix + "hello" + i).getBytes());
            //将event放入到list中
            list.add(event);
        }
        try {
            //将数据放入到channel中
            //获取channelProcessor
            ChannelProcessor channelProcessor = getChannelProcessor();
            channelProcessor.processEventBatch(list);
        } catch (Exception e) {
            status = Status.BACKOFF;
        }
        return status;
    }

    
    @Override
    public void configure(Context context) {
        //获取配置文件中的指定属性的内容，如果没有配置那么默认值为test
        prefix = context.getString("prefix", "test=");

    }

    
    @Override
    public long getBackOffSleepIncrement() {
        return 2000L;
    }

    
    @Override
    public long getMaxBackOffSleepInterval() {
        return 5000L;
    }
    
}

4、打包代码
使用xftp将包传入Linux中，传入flume目录下的lib目录中。

5、编写配置文件autosource.conf

a1.sources=r1
a1.channels=c1 c2
a1.sinks=k1 k2
#自定义source
a1.sources.r1.type=cn.jit.MySource

a1.channels.c1.type=memory

a1.sinks.k1.type=logger

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

6、执行命令
进入到flume目录下，执行命令flume-ng agent -n a1 -c conf/ -f datas2/autosource.conf -Dflume.root.logger=INFO,console

7、查看输出内容