1.問題背景

通過flume直接上傳實時資料到hdfs，會常遇到的一個問題就是小檔案，需要調引數來設定，往往在生產環境引數大小也不同

1.flume滾動配置為何不起作用？

2.通過原始碼分析得出什么原因？

3.該如何解決flume小檔案？

2. 程序分析

接著上一篇，https://blog.csdn.net/hu_lichao/article/details/110358689

本人在測驗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://linux121:9000/user/data/logs/%Y-%m-%d
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 是使用本地時間戳來對hdfs上的目錄來命名，這個屬性的目的就是相當于時間戳的攔截器，否則%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.sources = r1
a1.sinks = k1
a1.channels = c1
# taildir source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile =
/data/lagoudw/conf/startlog_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/hoult/servers/logs/start/.*log
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.hoult.flume.CustomerInterceptor$Builder
# memorychannel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 2000
# hdfs sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/data/logs/start/dt=%{logtime}/
a1.sinks.k1.hdfs.filePrefix = startlog.
# 組態檔滾動方式（檔案大小32M）
a1.sinks.k1.hdfs.rollSize = 33554432
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.idleTimeout = 0
a1.sinks.k1.hdfs.minBlockReplicas = 1
# 向hdfs上重繪的event的個數
a1.sinks.k1.hdfs.batchSize = 1000
# 使用本地時間
# a1.sinks.k1.hdfs.useLocalTimeStamp = true
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

這樣程式就永遠不會因為檔案所在塊的復制而滾動檔案了，只會根據你的配置項來滾動檔案了，，，，

3.總結

設定minBlockReplicas=1 的時候可以保證會按你設定的幾個引數來達到不會產生過多的小檔案，因為這個引數在讀取時候優先級較高，會首先判斷到有沒有Hdfs的副本復制，導致滾動檔案的例外，另外flume接入資料時候可以通過過濾器盡可能把一些完全用不到的資料進行過濾，清洗時候就省事一些了，
吳邪，小三爺，混跡于后臺，大資料，人工智能領域的小菜鳥，
更多請關注
file

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/230932.html

標籤：其他

上一篇：干貨收藏！639頁《深度學習：Deep Learning》圖文并茂課程PPT

下一篇：演算法-一個經典sql 題和一個Java演算法題

大資料開發-Flume-頻繁產生小檔案原因和處理

1.問題背景

2. 程序分析

3.總結