Flume快速入门系列(8) | 如何自定义Source

举报
不温卜火 发表于 2020/12/01 22:29:38 2020/12/01
【摘要】   这篇文章我们讲解的是如何自定义Source。 目录 1. Source的简单介绍2. 需求/分析3. 编码3.1 导入pom依赖3.2 编写代码 4. 测试 1. Source的简单介绍   Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据,包括avro、thrift、exec、...

  这篇文章我们讲解的是如何自定义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配置文件中配置。
1
2

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)下。
3
4

  • 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. 结果展示

展示的结果如预期一样,两秒一次
5
  本次的分享就到这里了,


11

   看 完 就 赞 , 养 成 习 惯 ! ! ! \color{#FF0000}{看完就赞,养成习惯!!!} ^ _ ^ ❤️ ❤️ ❤️
  码字不易,大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦!

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

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

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。