• Flume学习笔记（3）—— Flume 自定义组件

    前置知识：

    Flume学习笔记（1）—— Flume入门-CSDN博客

    Flume学习笔记（2）—— Flume进阶-CSDN博客

    Flume 自定义组件

    自定义 Interceptor

    需求分析：使用 Flume 采集服务器本地日志，需要按照日志类型的不同，将不同种类的日志发往不同的分析系统

    需要使用Flume 拓扑结构中的 Multiplexing 结构，Multiplexing的原理是，根据 event 中 Header 的某个 key 的值，将不同的 event 发送到不同的 Channel中，所以我们需要自定义一个 Interceptor，为不同类型的 event 的 Header 中的 key 赋予不同的值

    实现流程：

    代码

    导入依赖：

    1. <dependencies>
    2.   <dependency>
    3.     <groupId>org.apache.flumegroupId>
    4.     <artifactId>flume-ng-coreartifactId>
    5.     <version>1.9.0version>
    6.   dependency>
    7. dependencies>

    自定义拦截器的代码：

    1. package com.why.interceptor;
    2.  
    3. import org.apache.flume.Context;
    4. import org.apache.flume.Event;
    5. import org.apache.flume.interceptor.Interceptor;
    6.  
    7. import java.util.ArrayList;
    8. import java.util.List;
    9. import java.util.Map;
    10.  
    11. public class TypeInterceptor implements Interceptor {
    12.  
    13.     //存放事件的集合
    14.     private List addHeaderEvents;
    15.  
    16.     @Override
    17.     public void initialize() {
    18.         //初始化集合
    19.         addHeaderEvents = new ArrayList<>();
    20.     }
    21.  
    22.     //单个事件拦截
    23.     @Override
    24.     public Event intercept(Event event) {
    25.         //获取头信息
    26.         Map headers = event.getHeaders();
    27.         //获取body信息
    28.         String body = new String(event.getBody());
    29.  
    30.         //根据数据中是否包含”why“来分组
    31.         if(body.contains("why"))
    32.         {
    33.             headers.put("type","first");
    34.         }else {
    35.             headers.put("type","second");
    36.         }
    37.         return event;
    38.     }
    39.  
    40.     //批量事件拦截
    41.     @Override
    42.     public List intercept(List events) {
    43.         //清空集合
    44.         addHeaderEvents.clear();
    45.  
    46.         //遍历events
    47.         for(Event event : events)
    48.         {
    49.             //给每一个事件添加头信息
    50.             addHeaderEvents.add(intercept(event));
    51.         }
    52.         return addHeaderEvents;
    53.     }
    54.  
    55.     @Override
    56.     public void close() {
    57.  
    58.     }
    59.  
    60.     //构建生成器
    61.     public static class TypeBuilder implements Interceptor.Builder{
    62.  
    63.         @Override
    64.         public Interceptor build() {
    65.             return new TypeInterceptor();
    66.         }
    67.  
    68.         @Override
    69.         public void configure(Context context) {
    70.  
    71.         }
    72.     }
    73.  
    74. }

    将代码打包放入flume安装路径下的lib文件夹中

    配置文件

    在job文件夹下创建group4目录，添加配置文件；

    为 hadoop102 上的 Flume1 配置 1 个 netcat source，1 个 sink group（2 个 avro sink），并配置相应的 ChannelSelector 和 interceptor

    1. # Name the components on this agent
    2. a1.sources = r1
    3. a1.sinks = k1 k2
    4. a1.channels = c1 c2
    5.  
    6. # Describe/configure the source
    7. a1.sources.r1.type = netcat
    8. a1.sources.r1.bind = localhost
    9. a1.sources.r1.port = 44444
    10. a1.sources.r1.interceptors = i1
    11. a1.sources.r1.interceptors.i1.type = com.why.interceptor.TypeInterceptor$TypeBuilder
    12. a1.sources.r1.selector.type = multiplexing
    13. a1.sources.r1.selector.header = type
    14. a1.sources.r1.selector.mapping.first = c1
    15. a1.sources.r1.selector.mapping.second = c2
    16.  
    17. # Describe the sink
    18. a1.sinks.k1.type = avro
    19. a1.sinks.k1.hostname = hadoop103
    20. a1.sinks.k1.port = 4141
    21. a1.sinks.k2.type=avro
    22. a1.sinks.k2.hostname = hadoop104
    23. a1.sinks.k2.port = 4242
    24.  
    25. # Use a channel which buffers events in memory
    26. a1.channels.c1.type = memory
    27. a1.channels.c1.capacity = 1000
    28. a1.channels.c1.transactionCapacity = 100
    29. # Use a channel which buffers events in memory
    30. a1.channels.c2.type = memory
    31. a1.channels.c2.capacity = 1000
    32. a1.channels.c2.transactionCapacity = 100
    33.  
    34. # Bind the source and sink to the channel
    35. a1.sources.r1.channels = c1 c2
    36. a1.sinks.k1.channel = c1
    37. a1.sinks.k2.channel = c2

    hadoop103：配置一个 avro source 和一个 logger sink

    1. a1.sources = r1
    2. a1.sinks = k1
    3. a1.channels = c1
    4. a1.sources.r1.type = avro
    5. a1.sources.r1.bind = hadoop103
    6. a1.sources.r1.port = 4141
    7. a1.sinks.k1.type = logger
    8. a1.channels.c1.type = memory
    9. a1.channels.c1.capacity = 1000
    10. a1.channels.c1.transactionCapacity = 100
    11. a1.sinks.k1.channel = c1
    12. a1.sources.r1.channels = c1

    hadoop104：配置一个 avro source 和一个 logger sink

    1. a1.sources = r1
    2. a1.sinks = k1
    3. a1.channels = c1
    4. a1.sources.r1.type = avro
    5. a1.sources.r1.bind = hadoop104
    6. a1.sources.r1.port = 4242
    7. a1.sinks.k1.type = logger
    8. a1.channels.c1.type = memory
    9. a1.channels.c1.capacity = 1000
    10. a1.channels.c1.transactionCapacity = 100
    11. a1.sinks.k1.channel = c1
    12. a1.sources.r1.channels = c1

    执行指令

    hadoop102：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume-interceptor-flume.conf

    hadoop103：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume1-flume-logger.conf -Dflume.root.logger=INFO,console

    hadoop104：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume2-flume-logger.conf -Dflume.root.logger=INFO,console

    然后hadoop102通过nc连接44444端口，发送数据：

    在hadoop103和104上分别接收到：

    自定义 Source

    官方提供的文档：Flume 1.11.0 Developer Guide — Apache Flume

    给出的示例代码如下：

    1. public class MySource extends AbstractSource implements Configurable, PollableSource {
    2.   private String myProp;
    3.  
    4.   @Override
    5.   public void configure(Context context) {
    6.     String myProp = context.getString("myProp", "defaultValue");
    7.  
    8.     // Process the myProp value (e.g. validation, convert to another type, ...)
    9.  
    10.     // Store myProp for later retrieval by process() method
    11.     this.myProp = myProp;
    12.   }
    13.  
    14.   @Override
    15.   public void start() {
    16.     // Initialize the connection to the external client
    17.   }
    18.  
    19.   @Override
    20.   public void stop () {
    21.     // Disconnect from external client and do any additional cleanup
    22.     // (e.g. releasing resources or nulling-out field values) ..
    23.   }
    24.  
    25.   @Override
    26.   public Status process() throws EventDeliveryException {
    27.     Status status = null;
    28.  
    29.     try {
    30.       // This try clause includes whatever Channel/Event operations you want to do
    31.  
    32.       // Receive new data
    33.       Event e = getSomeData();
    34.  
    35.       // Store the Event into this Source's associated Channel(s)
    36.       getChannelProcessor().processEvent(e);
    37.  
    38.       status = Status.READY;
    39.     } catch (Throwable t) {
    40.       // Log exception, handle individual exceptions as needed
    41.  
    42.       status = Status.BACKOFF;
    43.  
    44.       // re-throw all Errors
    45.       if (t instanceof Error) {
    46.         throw (Error)t;
    47.       }
    48.     } finally {
    49.       txn.close();
    50.     }
    51.     return status;
    52.   }
    53. }

    需要继承AbstractSource，实现Configurable, PollableSource

    实战需求分析

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

    代码

    1. package com.why.source;
    2.  
    3. import org.apache.flume.Context;
    4. import org.apache.flume.EventDeliveryException;
    5. import org.apache.flume.PollableSource;
    6. import org.apache.flume.conf.Configurable;
    7. import org.apache.flume.event.SimpleEvent;
    8. import org.apache.flume.source.AbstractSource;
    9.  
    10. import java.util.HashMap;
    11. import java.util.concurrent.ConcurrentMap;
    12.  
    13. public class MySource extends AbstractSource implements PollableSource, Configurable {
    14.  
    15.     //定义配置文件将来要读取的字段
    16.     private Long delay;
    17.     private String field;
    18.  
    19.     //获取数据封装成 event 并写入 channel，这个方法将被循环调用
    20.     @Override
    21.     public Status process() throws EventDeliveryException {
    22.         try {
    23.             //事件头信息
    24.             HashMap headerMap = new HashMap<>();
    25.             //创建事件
    26.             SimpleEvent event = new SimpleEvent();
    27.             //循环封装事件
    28.             for (int i = 0; i < 5; i++) {
    29.                 //设置头信息
    30.                 event.setHeaders(headerMap);
    31.                 //设置事件内容
    32.                 event.setBody((field + i).getBytes());
    33.                 //将事件写入Channel
    34.                 getChannelProcessor().processEvent(event);
    35.                 Thread.sleep(delay);
    36.             }
    37.  
    38.         }catch (InterruptedException e) {
    39.             throw new RuntimeException(e);
    40.         }
    41.         return Status.READY;
    42.     }
    43.  
    44.     //backoff 步长
    45.     @Override
    46.     public long getBackOffSleepIncrement() {
    47.         return 0;
    48.     }
    49.  
    50.     //backoff 最长时间
    51.     @Override
    52.     public long getMaxBackOffSleepInterval() {
    53.         return 0;
    54.     }
    55.  
    56.     //初始化 context（读取配置文件内容）
    57.     @Override
    58.     public void configure(Context context) {
    59.         delay = context.getLong("delay");
    60.         field = context.getString("field","Hello");
    61.     }
    62.  
    63. }

    打包放到flume安装路径下的lib文件夹中；

    配置文件

    1. # Name the components on this agent
    2. a1.sources = r1
    3. a1.sinks = k1
    4. a1.channels = c1
    5.  
    6. # Describe/configure the source
    7. a1.sources.r1.type = com.why.source.MySource
    8. a1.sources.r1.delay = 1000
    9. a1.sources.r1.field = why
    10.  
    11. # Describe the sink
    12. a1.sinks.k1.type = logger
    13.  
    14. # Use a channel which buffers events in memory
    15. a1.channels.c1.type = memory
    16. a1.channels.c1.capacity = 1000
    17. a1.channels.c1.transactionCapacity = 100
    18.  
    19. # Bind the source and sink to the channel
    20. a1.sources.r1.channels = c1
    21. a1.sinks.k1.channel = c1

    执行指令

    hadoop102上：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group5/mysource.conf -Dflume.root.logger=INFO,console

    结果如下：

    自定义 Sink

    Sink 不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。

    Sink 是完全事务性的。在从 Channel 批量删除数据之前，每个 Sink 用 Channel 启动一个事务。批量事件一旦成功写出到存储系统或下一个 Flume Agent，Sink 就利用 Channel 提交事务。事务一旦被提交，该 Channel 从自己的内部缓冲区删除事件

    官方文档：Flume 1.11.0 Developer Guide — Apache Flume

    接口实例：

    1. public class MySink extends AbstractSink implements Configurable {
    2.     private String myProp;
    3.  
    4.     @Override
    5.     public void configure(Context context) {
    6.         String myProp = context.getString("myProp", "defaultValue");
    7.  
    8.         // Process the myProp value (e.g. validation)
    9.  
    10.         // Store myProp for later retrieval by process() method
    11.         this.myProp = myProp;
    12.     }
    13.  
    14.     @Override
    15.     public void start() {
    16.         // Initialize the connection to the external repository (e.g. HDFS) that
    17.         // this Sink will forward Events to ..
    18.     }
    19.  
    20.     @Override
    21.     public void stop () {
    22.         // Disconnect from the external respository and do any
    23.         // additional cleanup (e.g. releasing resources or nulling-out
    24.         // field values) ..
    25.     }
    26.  
    27.     @Override
    28.     public Status process() throws EventDeliveryException {
    29.         Status status = null;
    30.  
    31.         // Start transaction
    32.         Channel ch = getChannel();
    33.         Transaction txn = ch.getTransaction();
    34.         txn.begin();
    35.         try {
    36.             // This try clause includes whatever Channel operations you want to do
    37.  
    38.             Event event = ch.take();
    39.  
    40.             // Send the Event to the external repository.
    41.             // storeSomeData(e);
    42.  
    43.             txn.commit();
    44.             status = Status.READY;
    45.         } catch (Throwable t) {
    46.             txn.rollback();
    47.  
    48.             // Log exception, handle individual exceptions as needed
    49.  
    50.             status = Status.BACKOFF;
    51.  
    52.             // re-throw all Errors
    53.             if (t instanceof Error) {
    54.                 throw (Error)t;
    55.             }
    56.         }
    57.         return status;
    58.     }
    59. }

    自定义MySink 需要继承 AbstractSink 类并实现 Configurable 接口

    实战需求分析

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

    代码

    1. package com.why.sink;
    2.  
    3. import org.apache.flume.*;
    4. import org.apache.flume.conf.Configurable;
    5. import org.apache.flume.sink.AbstractSink;
    6.  
    7. import org.slf4j.Logger;
    8. import org.slf4j.LoggerFactory;
    9.  
    10. public class MySink extends AbstractSink implements Configurable {
    11.  
    12.     //创建 Logger 对象
    13.     private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class);
    14.     //前后缀
    15.     private String prefix;
    16.     private String suffix;
    17.  
    18.     @Override
    19.     public Status process() throws EventDeliveryException {
    20.         //声明返回值状态信息
    21.         Status status;
    22.         //获取当前 Sink 绑定的 Channel
    23.         Channel ch = getChannel();
    24.         //获取事务
    25.         Transaction txn = ch.getTransaction();
    26.         //声明事件
    27.         Event event;
    28.         //开启事务
    29.         txn.begin();
    30.         //读取 Channel 中的事件，直到读取到事件结束循环
    31.         while (true) {
    32.             event = ch.take();
    33.             if (event != null) {
    34.                 break;
    35.             }
    36.         }
    37.         try {
    38.             //处理事件（打印）
    39.             LOG.info(prefix + new String(event.getBody()) + suffix);
    40.             //事务提交
    41.             txn.commit();
    42.             status = Status.READY;
    43.         } catch (Exception e) {
    44.             //遇到异常，事务回滚
    45.             txn.rollback();
    46.             status = Status.BACKOFF;
    47.         } finally {
    48.             //关闭事务
    49.             txn.close();
    50.         }
    51.         return status;
    52.     }
    53.  
    54.     @Override
    55.     public void configure(Context context) {
    56.         prefix = context.getString("prefix", "hello");
    57.         suffix = context.getString("suffix");
    58.     }
    59. }

    打包放到flume安装路径下的lib文件夹中；

    配置文件

    1. # Name the components on this agent
    2. a1.sources = r1
    3. a1.sinks = k1
    4. a1.channels = c1
    5.  
    6. # Describe/configure the source
    7. a1.sources.r1.type = netcat
    8. a1.sources.r1.bind = localhost
    9. a1.sources.r1.port = 44444
    10.  
    11. # Describe the sink
    12. a1.sinks.k1.type = com.why.sink.MySink
    13. a1.sinks.k1.prefix = why:
    14. a1.sinks.k1.suffix = :why
    15.  
    16. # Use a channel which buffers events in memory
    17. a1.channels.c1.type = memory
    18. a1.channels.c1.capacity = 1000
    19. a1.channels.c1.transactionCapacity = 100
    20.  
    21. # Bind the source and sink to the channel
    22. a1.sources.r1.channels = c1
    23. a1.sinks.k1.channel = c1

    执行指令

    hadoop102上：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group6/mysink.conf -Dflume.root.logger=INFO,console

    结果如下：

  • 相关阅读:
    docker搭建maven私服
    Vue实现过渡动画的三种方式
    php 进程通信系列 （三）信号量
    malloc和new的本质区别
    promise async和await的方法与使用
    05 robotFrameWork+selenium2library 一维数组的使用
    es中的匹配显示问题
    ＜Linux＞ shell运行原理及Linux权限的理解
    【go学习合集】进阶数据类型2 --------＞切片
    go的netpoll学习
  • 原文地址：https://blog.csdn.net/qq_51235856/article/details/134475721