Flume技术之自定义Source

1. 自定义Source介绍

1. Source 是负责接收数据到 Flume Agent 的组件。Source 组件可以处理各种类型、各种格式的日志数据，包括 avro、thrift、exec、jms、spooling directory、netcat、sequence、generator、syslog、http、legacy。官方提供的 source 类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些 source。

2. 官方也提供了自定义 source 的接口：https://flume.apache.org/FlumeDeveloperGuide.html#source

3. 根据官方说明自定义MySource 需要继承 AbstractSource 类并实现 Configurable 和 PollableSource 接口。

   • 实现相应方法：
     • getBackOffSleepIncrement()//暂不用
     • getMaxBackOffSleepInterval()//暂不用
     • configure(Context context)//初始化 context（读取配置文件内容）
     • process()//获取数据封装成 event 并写入 channel，这个方法将被循环调用。
   • 使用场景：读取 MySQL 数据或者其他文件系统。

2. 需求

使用 flume 接收数据，并给每条数据添加前缀和后缀，输出到控制台。前缀可从 flume 配置文件中配置。

3. 自定义MySource代码

依赖：

        <dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.7.0</version></dependency>

代码部分：

/*** @Date 2020/8/11 16:18* @Version 10.21* @Author DuanChaojie*/
public class MySource extends AbstractSource implements Configurable, PollableSource {private String prefix;private String suffix;/*** 初始化 context（读取配置文件内容）* @param context*/public void configure(Context context) {prefix = context.getString("prefix", "feiji");suffix = context.getString("suffix", "xiaxian");}/*** 获取数据封装成 event 并写入 channel，这个方法将被循环调用。* @return* @throws EventDeliveryException*/public Status process() throws EventDeliveryException {Status  status = null;try {// 1.创建事件头信息Map<String,String> headers = new HashMap<String, String>();// 2.创建事件SimpleEvent event = new SimpleEvent();// 3.循环封装事件for (int i = 0; i < 6; i++) {// 3.1 给事件设置头信息event.setHeaders(headers);// 3.2 给事件设置内容event.setBody((prefix+i+suffix).getBytes());// 3.3 将事件写入channelgetChannelProcessor().processEvent(event);Thread.sleep(3000);status = Status.READY;}} catch (InterruptedException e) {e.printStackTrace();status = Status.BACKOFF;}return status;}public long getBackOffSleepIncrement() {return 0;}public long getMaxBackOffSleepInterval() {return 0;}
}

打成jar包guli-flume-1.0-SNAPSHOT.jar上传到/opt/module/flume/lib/目录下

4. 测试自定义的MySource

4.1 配置文件flume.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = com.atguigu.flume.source.MySource
a1.sources.r1.prefix = ll
a1.sources.r1.suffix = dd# 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.2 运行Flume

bin/flume-ng agent -c conf/ -n a1 -f job/mysource/flume.conf -Dflume.root.logger=INFO,console

4.3 查看结果

5. ChannelProcessor类中部分源码

  /*** Attempts to {@linkplain Channel#put(Event) put} the given events into each* configured channel. If any {@code required} channel throws a* {@link ChannelException}, that exception will be propagated.* <p>* <p>Note that if multiple channels are configured, some {@link Transaction}s* may have already been committed while others may be rolled back in the* case of an exception.** @param events A list of events to put into the configured channels.* @throws ChannelException when a write to a required channel fails.*/public void processEventBatch(List<Event> events) {Preconditions.checkNotNull(events, "Event list must not be null");events = interceptorChain.intercept(events);Map<Channel, List<Event>> reqChannelQueue =new LinkedHashMap<Channel, List<Event>>();Map<Channel, List<Event>> optChannelQueue =new LinkedHashMap<Channel, List<Event>>();for (Event event : events) {List<Channel> reqChannels = selector.getRequiredChannels(event);for (Channel ch : reqChannels) {List<Event> eventQueue = reqChannelQueue.get(ch);if (eventQueue == null) {eventQueue = new ArrayList<Event>();reqChannelQueue.put(ch, eventQueue);}eventQueue.add(event);}List<Channel> optChannels = selector.getOptionalChannels(event);for (Channel ch : optChannels) {List<Event> eventQueue = optChannelQueue.get(ch);if (eventQueue == null) {eventQueue = new ArrayList<Event>();optChannelQueue.put(ch, eventQueue);}eventQueue.add(event);}}// Process required channelsfor (Channel reqChannel : reqChannelQueue.keySet()) {Transaction tx = reqChannel.getTransaction();Preconditions.checkNotNull(tx, "Transaction object must not be null");try {tx.begin();List<Event> batch = reqChannelQueue.get(reqChannel);for (Event event : batch) {reqChannel.put(event);}tx.commit();} catch (Throwable t) {tx.rollback();if (t instanceof Error) {LOG.error("Error while writing to required channel: " + reqChannel, t);throw (Error) t;} else if (t instanceof ChannelException) {throw (ChannelException) t;} else {throw new ChannelException("Unable to put batch on required " +"channel: " + reqChannel, t);}} finally {if (tx != null) {tx.close();}}}// Process optional channelsfor (Channel optChannel : optChannelQueue.keySet()) {Transaction tx = optChannel.getTransaction();Preconditions.checkNotNull(tx, "Transaction object must not be null");try {tx.begin();List<Event> batch = optChannelQueue.get(optChannel);for (Event event : batch) {optChannel.put(event);}tx.commit();} catch (Throwable t) {tx.rollback();LOG.error("Unable to put batch on optional channel: " + optChannel, t);if (t instanceof Error) {throw (Error) t;}} finally {if (tx != null) {tx.close();}}}}

☆