0%

Flume自定义Sink

发表于 分类于
Sink 是完全事务性的。在从 Channel 批量删除数据之前，每个 Sink 用 Channel 启动一 个事务。批量事件一旦成功写出到存储系统或下一个 Flume Agent，Sink 就利用 Channel 提 交事务。事务一旦被提交，该 Channel 从自己的内部缓冲区删除事件。Sink 组件目的地包括 hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、 自定义。官方提供的 Sink 类型已经很多，但是有时候并不能满足实际开发当中的需求，此 时我们就需要根据实际需求自定义某些 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