【Flume】【源码分析】flume中sink到hdfs，文件系统频繁产生文件，文件滚动配置不起作用？

本人在测试hdfs的sink，发现sink端的文件滚动配置项起不到任何作用，配置如下：

a1.sinks.k1.type=hdfs
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.useLocalTimeStamp=true
a1.sinks.k1.hdfs.path=hdfs://192.168.11.177:9000/flume/events/%Y/%m/%d/%H/%M
a1.sinks.k1.hdfs.filePrefix=XXX
a1.sinks.k1.hdfs.rollInterval=60
a1.sinks.k1.hdfs.rollSize=0
a1.sinks.k1.hdfs.rollCount=0
a1.sinks.k1.hdfs.idleTimeout=0

这里配置的是60秒，文件滚动一次，也就每隔60秒，会新产生一个文件【前提，flume的source端有数据来】

这里注意

useLocalTimeStamp

这个属性的目的就是相当于时间戳的拦截器，否则%Y 等等这些东西都识别不了

要么用上面这个属性，要么用时间戳拦截器。

但是当我启动flume的时候，运行十几秒，不断写入数据，发现hdfs端频繁的产生文件，每隔几秒就有新文件产生
而且在flume的日志输出可以频繁看到这句：

[WARN] Block Under-replication detected. Rotating file.

只要有这句，就会产生一个新的文件

意思就是检测到复制块正在滚动文件，结合源码看下：

private boolean shouldRotate() {boolean doRotate = false;if (writer.isUnderReplicated()) {this.isUnderReplicated = true;doRotate = true;} else {this.isUnderReplicated = false;}if ((rollCount > 0) && (rollCount <= eventCounter)) {LOG.debug("rolling: rollCount: {}, events: {}", rollCount, eventCounter);doRotate = true;}if ((rollSize > 0) && (rollSize <= processSize)) {LOG.debug("rolling: rollSize: {}, bytes: {}", rollSize, processSize);doRotate = true;}return doRotate;}

这是判断是否滚动文件，但是这里面的第一判断条件是判断是否当前的HDFSWriter正在复制块

public boolean isUnderReplicated() {try {int numBlocks = getNumCurrentReplicas();if (numBlocks == -1) {return false;}int desiredBlocks;if (configuredMinReplicas != null) {desiredBlocks = configuredMinReplicas;} else {desiredBlocks = getFsDesiredReplication();}return numBlocks < desiredBlocks;} catch (IllegalAccessException e) {logger.error("Unexpected error while checking replication factor", e);} catch (InvocationTargetException e) {logger.error("Unexpected error while checking replication factor", e);} catch (IllegalArgumentException e) {logger.error("Unexpected error while checking replication factor", e);}return false;}

通过读取的配置复制块数量和当前正在复制的块比较，判断是否正在被复制

if (shouldRotate()) {boolean doRotate = true;if (isUnderReplicated) {if (maxConsecUnderReplRotations > 0 &&consecutiveUnderReplRotateCount >= maxConsecUnderReplRotations) {doRotate = false;if (consecutiveUnderReplRotateCount == maxConsecUnderReplRotations) {LOG.error("Hit max consecutive under-replication rotations ({}); " +"will not continue rolling files under this path due to " +"under-replication", maxConsecUnderReplRotations);}} else {LOG.warn("Block Under-replication detected. Rotating file.");}consecutiveUnderReplRotateCount++;} else {consecutiveUnderReplRotateCount = 0;}

以上方法，入口是shouldRotate()方法，也就是如果你配置了rollcount,rollsize大于0，会按照你的配置来滚动的，但是在入口进来后，发现，又去判断了是否有块在复制；

里面就读取了一个固定变量maxConsecUnderReplRotations=30，也就是正在复制的块，最多之能滚动出30个文件，如果超过了30次，该数据块如果还在复制中，那么数据也不会滚动了，doRotate=false，不会滚动了，所以有的人发现自己一旦运行一段时间，会出现30个文件

再结合上面的源码看一下：

如果你配置了10秒滚动一次，写了2秒，恰好这时候该文件内容所在的块在复制中，那么虽然没到10秒，依然会给你滚动文件的，文件大小，事件数量的配置同理了。

为了解决上述问题，我们只要让程序感知不到写的文件所在块正在复制就行了，怎么做呢？？

只要让isUnderReplicated()方法始终返回false就行了

该方法是通过当前正在被复制的块和配置中读取的复制块数量比较的，我们能改的就只有配置项中复制块的数量，而官方给出的flume配置项中有该项

hdfs.minBlockReplicas

Specify minimum number of replicas per HDFS block. If not specified, it comes from the default Hadoop config in the classpath.
默认读的是hadoop中的dfs.replication属性，该属性默认值是3

这里我们也不去该hadoop中的配置，在flume中添加上述属性为1即可

配置如下：

a1.sinks.k1.type=hdfs
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.useLocalTimeStamp=true
a1.sinks.k1.hdfs.path=hdfs://192.168.11.177:9000/flume/events/%Y/%m/%d/%H/%M
a1.sinks.k1.hdfs.filePrefix=cmcc
a1.sinks.k1.hdfs.minBlockReplicas=1
#a1.sinks.k1.hdfs.fileType=DataStream
#a1.sinks.k1.hdfs.writeFormat=Text
a1.sinks.k1.hdfs.rollInterval=60
a1.sinks.k1.hdfs.rollSize=0
a1.sinks.k1.hdfs.rollCount=0
a1.sinks.k1.hdfs.idleTimeout=0

这样程序就永远不会因为文件所在块的复制而滚动文件了，只会根据你的配置项来滚动文件了，试试吧！！

望各位网友不吝指教！！！

【Flume】【源码分析】flume中sink到hdfs，文件系统频繁产生文件，文件滚动配置不起作用？相关推荐

【SemiDrive源码分析】【Yocto源码分析】07 - core-image-base-x9h_ref_serdes.rootfs.ext4 文件系统是如何生成的
[SemiDrive源码分析][Yocto源码分析]07 - core-image-base-x9h_ref_serdes.rootfs.ext4 文件系统是如何生成的 1.core-image-ba ...
集合的get方法中参数从多少开始_源码分析CopyOnWriteArrayList 中的隐藏知识，你Get了吗？...
欢迎点击 "未读代码" ,关注公众号,文章每周更新杭州-阿里园区墙前言本觉 CopyOnWriteArrayList 过于简单,寻思看名字就能知道内部的实现逻辑,所以没有写这 ...
c+智能指针源码分析_C ++中的智能指针
c+智能指针源码分析 In this article, we'll take a look at how we can use Smart Pointers in C++. 在本文中,我们将研究如何在 ...
android 代码 drawable,Android Drawable完全解析（一）：Drawable源码分析（中）
呃...我不是故意要凑篇幅写个什么上下篇,实在是因为Drawable源码有点长,一篇写不下啦O(∩_∩)O~ 鉴于源码一般较长,以后所有源码分析的部分,英文注释非必要情况都不再保留! 2:Drawab ...
深入源码分析SpringBoot中使用@ConditionalOnBean无效的问题(@ConditionalOnBean did not find any beans of type)
一.前言最近在使用SpringBoot的@ConditionalOnBean的时候遇到一个很很奇特的问题.即在@Bean中使用@ConditionalOnBean注解,在可以确保需要依赖的Bean一 ...
源码分析 vue-cli 中安装依赖
最近想要自己实现一个 cli 看了 vue-cli 源码中安装依赖的部分源码文件 @vue/cli/lib/util/ProjectPackageManager.js 有一个 install 函数执 ...
【VUE】源码分析 - vue中的 HTMLParser，模板解析的第一步
tip:本系列博客的代码部分(示例等除外),均出自vue源码内容,版本为2.6.14.但是为了增加易读性,会对不相关内容做选择性省略.如果大家想了解完整的源码,建议自行从官方下载. GitHub ...
flume源码分析2--配置文件的加载
上面提到Application启动的时候,PollingPropertiesFileConfigurationProvider作为唯一的LifecycleAware类型的组件被交给监护者Lifecyc ...
java spring源码_spring源码分析-spring中的bean
接触过spring的人都知道,在spring中我们称java对象为bean,我们在spring的debug日志或者报错日志也能看到各种bean的描述.其实,spring的bean和java的对象之间是 ...
6 VPP源码分析 (VPP中的多线程)
VPP支持多线程模式,其中区分为main线程和worker线程,这两种线程都运行vlib_main_or_worker_loop函数作为线程主函数,主要区别在于执行时的is_main参数. 主线程可以 ...

【Flume】【源码分析】flume中sink到hdfs，文件系统频繁产生文件，文件滚动配置不起作用？

【Flume】【源码分析】flume中sink到hdfs，文件系统频繁产生文件，文件滚动配置不起作用？相关推荐

最新文章

热门文章