Flume-NG源码阅读之SpoolDirectorySource

2024-05-03 23:32
文章标签 源码 阅读 ng flume spooldirectorysource

本文主要是介绍Flume-NG源码阅读之SpoolDirectorySource，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

org.apache.flume.source.SpoolDirectorySource是flume的一个常用的source，这个源支持从 磁盘 中某 文件夹 获取文件数据。不同于其他异步源，这个源能够避免重启或者发送失败后数据丢失。flume可以 监控 文件夹，当出现新文件时会读取该文件并获取数据。当一个给定的文件被全部读入到通道中时，该文件会被重命名以标志已经完成。同时，该源需要一个清理进程来定期移除完成的文件。

　　通道可选地将一个完成路径的原始文件插入到每个事件的hearder域中。在读取文件时，source缓存文件数据到内存中。同时，需要确定设置了bufferMaxLineLength选项，以确保该数据远大于输入数据中数据最长的某一行。

     注意！！！channel只接收spooling directory中唯一命名的文件。如果文件名重复或文件在读取过程中被修改，则会有读取失败返回异常信息。这种场景下，同名的文件复制到这个目录时建议带唯一标示，比如时间戳。

     一、configure(Context context)方法。代码如下：
  1. public void configure(Context context) {
  2.     spoolDirectory = context.getString(SPOOL_DIRECTORY);
  3.     Preconditions.checkState(spoolDirectory != null,
  4.         "Configuration must specify a spooling directory");

  6.     completedSuffix = context.getString(SPOOLED_FILE_SUFFIX,
  7.         DEFAULT_SPOOLED_FILE_SUFFIX);
  8.     deletePolicy = context.getString(DELETE_POLICY, DEFAULT_DELETE_POLICY);
  9.     fileHeader = context.getBoolean(FILENAME_HEADER,
  10.         DEFAULT_FILE_HEADER);
  11.     fileHeaderKey = context.getString(FILENAME_HEADER_KEY,
  12.         DEFAULT_FILENAME_HEADER_KEY);
  13.     batchSize = context.getInteger(BATCH_SIZE,
  14.         DEFAULT_BATCH_SIZE);
  15.     inputCharset = context.getString(INPUT_CHARSET, DEFAULT_INPUT_CHARSET);

  17.     ignorePattern = context.getString(IGNORE_PAT, DEFAULT_IGNORE_PAT);
  18.     trackerDirPath = context.getString(TRACKER_DIR, DEFAULT_TRACKER_DIR);

  20.     deserializerType = context.getString(DESERIALIZER, DEFAULT_DESERIALIZER);
  21.     deserializerContext = new Context(context.getSubProperties(DESERIALIZER +
  22.         "."));

  24.     // "Hack" to support backwards compatibility with previous generation of
  25.     // spooling directory source, which did not support deserializers
  26.     Integer bufferMaxLineLength = context.getInteger(BUFFER_MAX_LINE_LENGTH);
  27.     if (bufferMaxLineLength != null && deserializerType != null &&
  28.         deserializerType.equals(DEFAULT_DESERIALIZER)) {
  29.       deserializerContext.put(LineDeserializer.MAXLINE_KEY,
  30.           bufferMaxLineLength.toString());
  31.     }

  33.   }
复制代码

　　1、spoolDirectory是 监控 目录，不能为空，没有默认值。这个source不具有监控子目录的功能，也就是不能递归监控。如果需要，这需要自己去实现， Flume-ng-1.3.0 spooling source的方式增加了对目录的递归检测的支持 这里有递归 检测 的实现；

　　2、completedSuffix是文件读取完毕后给完成文件添加的标记后缀，默认是".COMPLETED"；

　　3、deletePolicy这是是否删除读取完毕的文件，默认是"never"，就是不删除，目前只支持"never"和“IMMEDIATE”；

　　4、fileHeader是否在event的Header中添加文件名，boolean类型

　　5、fileHeaderKey这是event的Header中的key,value是文件名

　　6、batchSize这个是一次处理的记录数，默认是100；

　　7、inputCharset编码方式，默认是"UTF-8"；

　　8、ignorePattern忽略符合条件的文件名

　　9、trackerDirPath被处理文件元数据的存储目录，默认".flumespool"

　　10、deserializerType将文件中的数据序列化成event的方式，默认是“LINE”---org.apache.flume.serialization.LineDeserializer

　　11、deserializerContext这个主要用在Deserializer中设置编码方式outputCharset和文件每行最大长度maxLineLength。



　　二、start()方法。代码如下：
  1. public void start() {
  2.     logger.info("SpoolDirectorySource source starting with directory: {}",
  3.         spoolDirectory);

  5.     ScheduledExecutorService executor =
  6.         Executors.newSingleThreadScheduledExecutor();
  7.     counterGroup = new CounterGroup();

  9.     File directory = new File(spoolDirectory);
  10.     try {
  11.       reader = new ReliableSpoolingFileEventReader.Builder()
  12.           .spoolDirectory(directory)
  13.           .completedSuffix(completedSuffix)
  14.           .ignorePattern(ignorePattern)
  15.           .trackerDirPath(trackerDirPath)
  16.           .annotateFileName(fileHeader)
  17.           .fileNameHeader(fileHeaderKey)
  18.           .deserializerType(deserializerType)
  19.           .deserializerContext(deserializerContext)
  20.           .deletePolicy(deletePolicy)
  21.           .inputCharset(inputCharset)
  22.           .build();
  23.     } catch (IOException ioe) {
  24.       throw new FlumeException("Error instantiating spooling event parser",
  25.           ioe);
  26.     }

  28.     Runnable runner = new SpoolDirectoryRunnable(reader, counterGroup);
  29.     executor.scheduleWithFixedDelay(
  30.         runner, 0, POLL_DELAY_MS, TimeUnit.MILLISECONDS);

  32.     super.start();
  33.     logger.debug("SpoolDirectorySource source started");
  34.   }
复制代码

　　1、构建了一个org.apache.flume.client.avro.ReliableSpoolingFileEventReader的对象reader；

　　2、启动了一个每隔POLL_DELAY_MS（默认500，单位ms）执行一次SpoolDirectoryRunnable的进程；

　 　三、读取并发送event进程。代码如下：
  1. private class SpoolDirectoryRunnable implements Runnable {
  2.     private ReliableSpoolingFileEventReader reader;
  3.     private CounterGroup counterGroup;

  5.     public SpoolDirectoryRunnable(ReliableSpoolingFileEventReader reader,
  6.         CounterGroup counterGroup) {
  7.       this.reader = reader;
  8.       this.counterGroup = counterGroup;
  9.     }

  11.     @Override
  12.     public void run() {
  13.       try {
  14.         while (true) {
  15.           List<Event> events = reader.readEvents(batchSize);　　//读取batchSize个记录
  16.           if (events.isEmpty()) {
  17.             break;
  18.           }
  19.           counterGroup.addAndGet("spooler.events.read", (long) events.size());

  21.           getChannelProcessor().processEventBatch(events);　　//将events批量发送到channel
  22.           reader.commit();
  23.         }
  24.       } catch (Throwable t) {
  25.         logger.error("Uncaught exception in Runnable", t);
  26.         if (t instanceof Error) {
  27.           throw (Error) t;
  28.         }
  29.       }
  30.     }
  31.   }
复制代码

该进程实现了批量读取reader所指向的文件的数据，并发送到channel。

四、org.apache.flume.client.avro.ReliableSpoolingFileEventReader的构造方法首先是先尝试对spoolDirectory是否有创建文件、读、写、删除等权限；然后在构造"$spoolDirectory/.flumespool/.flumespool-main.meta"元数据文件

五、上面SpoolDirectoryRunnable.run方法中的List<Event> events = reader.readEvents(batchSize)，是org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(batchSize)：
  1. public List<Event> readEvents(int numEvents) throws IOException {
  2.     if (!committed) {
  3.       if (!currentFile.isPresent()) {//为空,如果Optional包含非null的引用（引用存在），返回true
  4.         throw new IllegalStateException("File should not roll when " +
  5.             "commit is outstanding.");
  6.       }
  7.       logger.info("Last read was never committed - resetting mark position.");
  8.       currentFile.get().getDeserializer().reset();
  9.     } else {//已经committed成功
  10.       // Check if new files have arrived since last call
  11.       //Returns true if this holder contains a (non-null) instance
  12.       if (!currentFile.isPresent()) {//为空，获取下一个文件，初次调用
  13.         currentFile = getNextFile();
  14.       }
  15.       // Return empty list if no new files
  16.       if (!currentFile.isPresent()) {//为空，已经没有可读的文件了
  17.         return Collections.emptyList();
  18.       }
  19. 　　　　//其它的说明是currentFile目前还在读
  20.     }

  22.     EventDeserializer des = currentFile.get().getDeserializer();
  23.     List<Event> events = des.readEvents(numEvents);//添加event的body

  25.     /* It's possible that the last read took us just up to a file boundary.
  26.      * If so, try to roll to the next file, if there is one. */
  27.     if (events.isEmpty()) {
  28.       retireCurrentFile();  //改名字
  29.       currentFile = getNextFile();//换下一个文件
  30.       if (!currentFile.isPresent()) {
  31.         return Collections.emptyList();
  32.       }
  33.       events = currentFile.get().getDeserializer().readEvents(numEvents);//继续读,添加event的body
  34.     }

  36.     if (annotateFileName) {
  37.       String filename = currentFile.get().getFile().getAbsolutePath();
  38.       for (Event event : events) {
  39.         event.getHeaders().put(fileNameHeader, filename);//添加header
  40.       }
  41.     }

  43.     committed = false;
  44.     lastFileRead = currentFile;
  45.     return events;
  46.   }<span style="background-color: rgb(255, 255, 255); line-height: 1.5;"> </span>
复制代码


1，committed初始化时是true，所以第一次运行就是通过getNextFile()获取当前要去读的文件。如果是空就返回空值了。

2，使用deserializer（默认是org.apache.flume.serialization.LineDeserializer）的readEvents(numEvents)去批量读数据封装成event。

3，如获取的批量events为空，说明这个文件读完了，需要对这个读完的文件做个“删除”（retireCurrentFile()方法，在这也会删除元数据文件），就是根据deletePolicy（删除还是添加去读完毕后缀completedSuffix）；但是这个本方法是有返回值的就是events，所以需要获取下一个文件，即再次运行getNextFile()，并events = currentFile.get().getDeserializer().readEvents(numEvents)

4，是否要对这些events的Header中添加文件名

5，committed = false;    lastFileRead = currentFile; 并返回events。

这个方法还有几点需要解释：

其一、就是committed参数，此参数关系到这一批量的event是否已经正确处理完毕。可以看到上面的5中所讲，每调用一次ReliableSpoolingFileEventReader.readEvents(batchSize)均会在最后将committed设置为false，但是在SpoolDirectoryRunnable.run()方法中也可以看出在调用readEvents方法后还会调用ReliableSpoolingFileEventReader.commit()方法，代码如下：

  1. /** Commit the last lines which were read. */
  2.   @Override
  3.   public void commit() throws IOException {
  4.     if (!committed && currentFile.isPresent()) {
  5.       currentFile.get().getDeserializer().mark();
  6.       committed = true;
  7.     }
复制代码


这个方法说明满足两个条件就可以：一、向trackerFile写入读到的记录位置，mark()方法会将syncPosition写入trackerFile，而ResettableFileInputStream中的position用来暂存位置增加的，待到何时会syncPosition=position，这样是为了防止出现异常时用于恢复丢失的数据；二、将committed  = true。两个条件：一个是committed=false，这个执行完readEvents最后会置为false；二、currentFile“非空”，代表有正在读的文件。如果committed在readEvents中开始时为false，说明：一、event提交到channel时出现了问题，没有执行reader.commit；二、currentFile已经“为空”，说明没有可以读的文件。这两点也体现在readEvents开始部分，committed=false时，如果没有可读文件就会抛出异常File should not roll when commit is outstanding."；如果是在提交到channel时出问题会通过currentFile.get().getDeserializer().reset()重新撤回到上次正确提交channel的位置，这样可以使得不丢失数据。

其二、就是getNextFile()方法。这个方法会首先过滤 检测 目录的子目录（也就是不能递归）、隐藏文件(以"."开头的文件)、已经读完的文件（有completedSuffix后缀的）、符合ignorePattern的文件；然后将过滤后的文件按时间的先后顺序排序，再创建一个新的对应的元数据文件；构造一个读取文件的输入流ResettableFileInputStream，并将此输入流作为参数传递给deserializer，最终返回一个Optional.of(new FileInfo(nextFile, deserializer));

其三、就是LineDeserializer）的readEvents(numEvents)方法。这个方法会多次(numEvents)调用LineDeserializer（默认）的readLine()获取一行数据封装成event。readLine()会通过org.apache.flume.serialization.ResettableFileInputStream.readChar()不断的去获取数据，读完正行后判断每行的长度是否超过规定值maxLineLength。readChar()方法除了不断读取一个字符外，还会记下字符的位置，等待将位置写入元数据文件中(通过deserializer.mark()写入)

这篇关于Flume-NG源码阅读之SpoolDirectorySource的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/957891。

相关文章

Go中sync.Once源码的深度讲解

Go中sync.Once源码的深度讲解

《Go中sync.Once源码的深度讲解》sync.Once是Go语言标准库中的一个同步原语,用于确保某个操作只执行一次,本文将从源码出发为大家详细介绍一下sync.Once的具体使用,x希望对大家有... 目录概念简单示例源码解读总结概念sync.Once是Go语言标准库中的一个同步原语，用于确保某个操
阅读更多...
Java汇编源码如何查看环境搭建

Java汇编源码如何查看环境搭建

《Java汇编源码如何查看环境搭建》：本文主要介绍如何在IntelliJIDEA开发环境中搭建字节码和汇编环境,以便更好地进行代码调优和JVM学习,首先,介绍了如何配置IntelliJIDEA以方... 目录一、简介二、在IDEA开发环境中搭建汇编环境2.1 在IDEA中搭建字节码查看环境2.1.1 搭建步
阅读更多...
JAVA智听未来一站式有声阅读平台听书系统小程序源码

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来，一站式有声阅读平台听书系统 🌟&nbsp;开篇：遇见未来，从“智听”开始 在这个快节奏的时代，你是否渴望在忙碌的间隙，找到一片属于自己的宁静角落？是否梦想着能随时随地，沉浸在知识的海洋，或是故事的奇幻世界里？今天，就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统，它正悄悄改变着我们的阅读方式，让未来触手可及！ 📚&nbsp;第一站：海量资源，应有尽有 走进“智听
阅读更多...
Java ArrayList扩容机制 （源码解读）

Java ArrayList扩容机制 （源码解读）

结论：初始长度为10，若所需长度小于1.5倍原长度，则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1：数组默认长度         2:这是一个共享的空数组实例，用于明确创建长度为0时的ArrayList ，比如通过 new ArrayList<>(0)，ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL
阅读更多...
如何在Visual Studio中调试.NET源码

如何在Visual Studio中调试.NET源码

今天偶然在看别人代码时，发现在他的代码里使用了Any判断List<T>是否为空。 我一般的做法是先判断是否为null，再判断Count。 看了一下Count的源码如下： 1 [__DynamicallyInvokable]2 public int Count3 {4 [__DynamicallyInvokable]5 get
阅读更多...
工厂ERP管理系统实现源码（JAVA）

工厂ERP管理系统实现源码（JAVA）

工厂进销存管理系统是一个集采购管理、仓库管理、生产管理和销售管理于一体的综合解决方案。该系统旨在帮助企业优化流程、提高效率、降低成本，并实时掌握各环节的运营状况。 在采购管理方面，系统能够处理采购订单、供应商管理和采购入库等流程，确保采购过程的透明和高效。仓库管理方面，实现库存的精准管理，包括入库、出库、盘点等操作，确保库存数据的准确性和实时性。 生产管理模块则涵盖了生产计划制定、物料需求计划、
阅读更多...
flume系列之：查看flume系统日志、查看统计flume日志类型、查看flume日志

flume系列之：查看flume系统日志、查看统计flume日志类型、查看flume日志

遍历指定目录下多个文件查找指定内容 服务器系统日志会记录flume相关日志 cat /var/log/messages |grep -i oom 查找系统日志中关于flume的指定日志 import osdef search_string_in_files(directory, search_string):count = 0
阅读更多...
论文阅读笔记: Segment Anything

论文阅读笔记: Segment Anything

文章目录 Segment Anything摘要引言任务模型数据引擎数据集负责任的人工智能 Segment Anything Model图像编码器提示编码器mask解码器解决歧义损失和训练 Segment Anything 论文地址: https://arxiv.org/abs/2304.02643 代码地址:https://github.com/facebookresear
阅读更多...
Spring 源码解读：自定义实现Bean定义的注册与解析

Spring 源码解读：自定义实现Bean定义的注册与解析

引言 在Spring框架中，Bean的注册与解析是整个依赖注入流程的核心步骤。通过Bean定义，Spring容器知道如何创建、配置和管理每个Bean实例。本篇文章将通过实现一个简化版的Bean定义注册与解析机制，帮助你理解Spring框架背后的设计逻辑。我们还将对比Spring中的BeanDefinition和BeanDefinitionRegistry，以全面掌握Bean注册和解析的核心原理。
阅读更多...
音视频入门基础：WAV专题（10）——FFmpeg源码中计算WAV音频文件每个packet的pts、dts的实现

音视频入门基础：WAV专题（10）——FFmpeg源码中计算WAV音频文件每个packet的pts、dts的实现

一、引言 从文章《音视频入门基础：WAV专题（6）——通过FFprobe显示WAV音频文件每个数据包的信息》中我们可以知道，通过FFprobe命令可以打印WAV音频文件每个packet（也称为数据包或多媒体包）的信息，这些信息包含该packet的pts、dts： 打印出来的“pts”实际是AVPacket结构体中的成员变量pts，是以AVStream->time_base为单位的显
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2