这篇文章我们讲解的是如何自定义Source。

1. 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（读取配置文件内容）

  

 

  
1
  
2
  
3
 

process()//获取数据封装成event并写入channel，这个方法将被循环调用。
使用场景：读取MySQL数据或者其他文件系统。

  

 

  
1
  
2
 

2. 需求/分析

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

3. 编码

3.1 导入pom依赖

 <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>8</source> <target>8</target> </configuration> </plugin> </plugins> </build> <dependencies> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.7.0</version> </dependency> </dependencies>


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
 

3.2 编写代码

package com.buwenbuhuo;


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;
import java.util.Map;

/**
 * @author 卜温不火
 * @create 2020-05-04 16:00
 * com.buwenbuhuo - the name of the target package where the new class or interface will be created.
 * flumeplugins0504 - the name of the current project.
 */

public class MySource extends AbstractSource implements Configurable, PollableSource { //定义需要从配置中读取的字段 //两条数据之间的间隔 private long delay; private String field; public Status process() throws EventDeliveryException { try { Map<String, String> header = new HashMap<>(); SimpleEvent event = new SimpleEvent(); //拿到数据 for (int i = 0; i < 5; i++) { event.setHeaders(header); event.setBody((field + i).getBytes()); getChannelProcessor().processEvent(event); Thread.sleep(delay); } } catch (Exception e) { return Status.BACKOFF; } return Status.READY; } public long getBackOffSleepIncrement() { return 0; } public long getMaxBackOffSleepInterval() { return 0; } public void configure(Context context) { delay = context.getLong("delay", 2000l); field = context.getString("field", "buwenbuhuo"); }
}

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
  
42
  
43
  
44
  
45
  
46
  
47
  
48
  
49
  
50
  
51
  
52
  
53
  
54
  
55
  
56
  
57
  
58
  
59
 

4. 测试

• 1 打包上传

将写好的代码打包，并放到flume的lib目录（/opt/module/flume/lib）下。

• 2. 配置文件

[bigdata@hadoop002 job]$ cp flume-netcat-logger.conf flume-mysource-logger.conf
[bigdata@hadoop002 job]$ vim flume-mysource-logger.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.buwenbuhuo.MySource
a1.sources.r1.delay = 2000
a1.sources.r1.field = buwenbuhuo

# 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


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
 

• 3. 开启任务

[bigdata@hadoop002 flume]$ [bigdata@hadoop002 flume]$ bin/flume-ng agent -c conf/ -f job/flume-mysource-logger.conf  -n a1 -Dflume.root.logger=INFO,console


  

 

  
1
  
2
 

• 4. 结果展示

展示的结果如预期一样，两秒一次

  本次的分享就到这里了,

  $看完就赞，养成习惯！！！$^ _ ^ ❤️ ❤️ ❤️
  码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！

文章来源: buwenbuhuo.blog.csdn.net，作者：不温卜火，版权归原作者所有，如需转载，请联系作者。

原文链接：buwenbuhuo.blog.csdn.net/article/details/105916648