0%

Flume自定义Sink

本文为尚硅谷Flume课程随堂笔记



一、 项目介绍

使用 flume 接收数据,并在 Sink 端给每条数据添加前缀和后缀,输出到控制台。前后 缀可在flume任务配置文件中配置。

https://flume.apache.org/FlumeDeveloperGuide.html#sink 根据官方说明自定义 MySink 需要继承 AbstractSink 类并实现 Configurable 接口。



二、 Sink

1.创建maven项目

1
2
3
4
5
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.7.0</version>
</dependency>

2. 构建自定义类

根据官方说明自定义 MySink 需要继承 AbstractSink 类并实现 Configurable 接口。
具体代码如下:

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
60
61
62
63
64
65
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

/**
* @author suiwo
* @title MySink
* @date 2020/6/2 14:29
* @description //todo
*/
public class MySink extends AbstractSink implements Configurable {

private String prefix;
private String subfix;

/**
* 1. 获取Channel
* 2. 从Channel获取事务以及数据
* 3. 发送数据
*/
@Override
public Status process() throws EventDeliveryException {

// 1. 定义返回值
Status status;

// 2. 获取Channel
Channel channel = getChannel();

// 3. 从Channel中获取事务
Transaction transaction = channel.getTransaction();

// 4. 启动事务
transaction.begin();
Event event;
do {
// 5. 从Channel中获取数据
event = channel.take();
} while (event == null);

try {
// 6. 处理事件
String body = new String(event.getBody());
System.out.println(prefix + "--" + body + "--" + subfix);

// 7. 提交事务
transaction.commit();
status = Status.READY;
} catch (ChannelException e) {
transaction.rollback();
// 10. 修改状态
status = Status.BACKOFF;
} finally {
// 11. 关闭事务
transaction.close();
}
return status;
}

@Override
public void configure(Context context) {
prefix = context.getString("prefix");
subfix = context.getString("subfix");
}
}


3. maven打包

将项目打包,并将项目放至flume的lib目录下



三、编写Flume相关配置

1. 编写my-sink.conf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = xyz.suiwo.flume.sink.MySink
a1.sinks.k1.prefix = suiwo
a1.sinks.k1.subfix = haha

# 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
bin/flume-ng agent -c conf -f job/my-sink.conf -n a1 -Dflume.root.logger=INFO,console