Flume开发 -- 自定义Source flume

一、自定义 Source 介绍 Source 是负责接收数据到 Flume Agent 的组件。Source 组件可以处理各种类型、各种格式的日志数据，包括 avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。官方提供的 source 类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些 source。
官方也提供了自定义 source 的接口：
https://flume.apache.org/FlumeDeveloperGuide.html#source
根据官方说明自定义 MySource 需要继承 AbstractSource 类并实现 Configurable 和 PollableSource 接口。
实现相应方法：
getBackOffSleepIncrement() //暂不用
getMaxBackOffSleepInterval() //暂不用
configure(Context context) //初始化 context（读取配置文件内容）
process() //获取数据封装成 event 并写入 channel，这个方法将被循环调用。
使用场景：读取 MySQL 数据或者其他文件系统。
二、需求使用 flume 接收数据，并给每条数据添加前缀，输出到控制台。前缀可以从 flume 配置文件中配置。

文章图片

三、分析
文章图片

四、案例实操 4.1 环境配置
创建 maven 项目，添加 pom 依赖。

org.apache.flume flume-ng-core 1.7.0

4.2 编写代码

package customersource; 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; import java.util.HashMap; /** * @description: 自定义 Source * @author: hyr * @time: 2020/3/4 8:27 */ public class MySource extends AbstractSource implements Configurable, PollableSource {// 定义配置文件将来要读取的字段 private Long delay; private String field; // 初始化配置信息 @Override public void configure(Context context) {// 无默认值 delay = context.getLong("delay"); // 有默认值 field = context.getString("field", "Hello!"); }/* 1、接收数据(for 循环造数据) 2、封装为事件 3、将事件传给 channel */ @Override public Status process() throws EventDeliveryException {try {// 创建事件头信息 HashMap, String> headerMap = new HashMap<>(); // 创建事件 SimpleEvent event = new SimpleEvent(); // 循环封装事件 for (int i = 0; i < 5; i++) {// 在源码中，SimpleEvent 的构造函数中，初始化了一个 HashMap。 // 若我们需要传入头信息，需要自己封装一个 HashMap。 event.setHeaders(headerMap); // 给事件设置内容 // 要传入一个字节数组 event.setBody((field + i).getBytes()); // 将事件写入 channel getChannelProcessor().processEvent(event); // 设置休眠时间 Thread.sleep(delay); } } catch (Exception e) {e.printStackTrace(); return Status.BACKOFF; }// 返回状态 return Status.READY; }@Override public long getBackOffSleepIncrement() {return 0; }@Override public long getMaxBackOffSleepInterval() {return 0; } }

4.3 打包
【Flume开发 -- 自定义Source】用 maven 将代码打成 jar 包，上传到服务器的 flume 的 lib 目录下。
4.4 编写 mySource.conf

# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = customersource.MySource a1.sources.r1.delay = 1000 # a1.sources.r1.field = helloworld # Describe the sink a1.sinks.k1.type = logger # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1

4.5 开启 Flume