大数据技术之Flume 企业开发案例—

大数据技术之Flume 企业开发案例——自定义 Source（9）

news2024/9/23 7:22:36

自定义 Source

1）介绍

2）需求

3）分析

4）编码

5）测试

自定义 Source

1）介绍

Source 是负责接收数据到 Flume Agent 的组件。Source 组件可以处理各种类型和格式的日志数据，包括 avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy 等。虽然官方提供的 Source 类型已经很丰富，但在实际开发中可能仍不能完全满足需求。此时，可以根据实际需求来自定义 Source。

官方提供了自定义 Source 的接口：Flume Developer Guidehttps://flume.apache.org/FlumeDeveloperGuide.html#source。自定义 MySource 需要继承 AbstractSource 类并实现 Configurable 和 PollableSource 接口。

主要实现的方法包括：

getBackOffSleepIncrement() —— backoff 步长
getMaxBackOffSleepInterval() —— backoff 最长时间
configure(Context context) —— 初始化 context（读取配置文件内容）
process() —— 获取数据封装成 event 并写入 channel，这个方法将被循环调用。

使用场景：例如读取 MySQL 数据或其他文件系统。

2）需求

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

3）分析

自定义 Source 需求分析：

继承 AbstractSource
实现 Configurable 和 PollableSource

关键方法：

configure(Context context)：读取配置文件(如 XX.conf)中的配置信息
process()：接收数据，将数据封装成一个个的 Event，写入 Channel。使用 for 循环模拟数据生成（例如：for(int i = 0; i < 5; i++)）
getBackOffSleepIncrement()：暂不使用
getMaxBackOffSleepInterval()：暂不使用

4）编码

（1）导入 pom 依赖

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

（2）编写代码

package com.lzl;

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;

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!");
  }

  @Override
  public Status process() throws EventDeliveryException {
    try {
      HashMap<String, String> headerMap = new HashMap<>();
      SimpleEvent event = new SimpleEvent();
      for (int i = 0; i < 5; i++) {
        event.setHeaders(headerMap);
        event.setBody((field + i).getBytes());
        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;
  }
}

5）测试

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

（2）配置文件

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = com.lzl.MySource
a1.sources.r1.delay = 1000
#a1.sources.r1.field = lzl

# 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

（3）开启任务

[lzl@hadoop12 flume]$ pwd
/opt/module/flume
[lzl@hadoop12 flume]$ bin/flume-ng agent -c conf/ -f job/mysource.conf -n a1 -Dflume.root.logger=INFO,console

（4）结果展示