SparkStreaming如何解決小文件問題

這期內容當中小編將會給大家?guī)碛嘘PSparkStreaming如何解決小文件問題，文章內容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

使用sparkstreaming時，如果實時計算結果要寫入到HDFS，那么不可避免的會遇到一個問題，那就是在默認情況下會產生非常多的小文件，這是由sparkstreaming的微批處理模式和DStream(RDD)的分布式(partition)特性導致的，sparkstreaming為每個partition啟動一個獨立的線程來處理數據，一旦文件輸出到HDFS，那么這個文件流就關閉了，再來一個batch的parttition任務，就再使用一個新的文件流，那么假設，一個batch為10s，每個輸出的DStream有32個partition，那么一個小時產生的文件數將會達到(3600/10)*32=11520個之多。眾多小文件帶來的結果是有大量的文件元信息，比如文件的location、文件大小、block number等需要NameNode來維護，NameNode會因此鴨梨山大。不管是什么格式的文件，parquet、text,、JSON或者 Avro，都會遇到這種小文件問題，這里討論幾種處理Sparkstreaming小文件的典型方法。

這種方法很容易理解，batch越大，從外部接收的event就越多，內存積累的數據也就越多，那么輸出的文件數也就回變少，比如上邊的時間從10s增加為100s，那么一個小時的文件數量就會減少到1152個。但別高興太早，實時業(yè)務能等那么久嗎，本來人家10s看到結果更新一次，現(xiàn)在要等快兩分鐘，是人都會罵娘。所以這種方法適用的場景是消息實時到達，但不想擠壓在一起處理，因為擠壓在一起處理的話，批處理任務在干等，這時就可以采用這種方法(是不是很像spark內部的pipeline模式，但是要注意區(qū)別哦)。

文章開頭講了，小文件的基數是：batch_number*partition_number，而第一種方法是減少batch_number，那么這種方法就是減少partition_number了，這個api不細說，就是減少初始的分區(qū)個數?？催^spark源碼的童鞋都知道，對于窄依賴，一個子RDD的partition規(guī)則繼承父RDD，對于寬依賴(就是那些個叉叉叉ByKey操作)，如果沒有特殊指定分區(qū)個數，也繼承自父rdd。那么初始的SourceDstream是幾個partiion，最終的輸出就是幾個partition。所以Coalesce大法的好處就是，可以在最終要輸出的時候，來減少一把partition個數。但是這個方法的缺點也很明顯，本來是32個線程在寫256M數據，現(xiàn)在可能變成了4個線程在寫256M數據，而沒有寫完成這256M數據，這個batch是不算做結束的。那么一個batch的處理時延必定增長，batch擠壓會逐漸增大。這種方法也要慎用，切雞切雞?。?br/>

我們既然把數據輸出到hdfs，那么說明肯定是要用hive或者sparksql這樣的“sql on hadoop”系統(tǒng)類進一步進行數據分析，而這些表一般都是按照半小時或者一小時、一天，這樣來分區(qū)的(注意不要和sparkStreaming的分區(qū)混淆，這里的分區(qū)，是用來做分區(qū)裁剪優(yōu)化的)，那么我們可以考慮在SparkStreaming外再啟動定時的批處理任務來合并SparkStreaming產生的小文件。這種方法不是很直接，但是卻比較有用，“性價比”較高，唯一要注意的是，批處理的合并任務在時間切割上要把握好，搞不好就可能回去合并一個還在寫入的SparkStreaming小文件。

SparkStreaming提供的foreach這個outout類api，可以讓我們自定義輸出計算結果的方法。那么我們其實也可以利用這個特性，那就是每個batch在要寫文件時，并不是去生成一個新的文件流，而是把之前的文件打開。考慮這種方法的可行性，首先，HDFS上的文件不支持修改，但是很多都支持追加，那么每個batch的每個partition就對應一個輸出文件，每次都去追加這個partition對應的輸出文件，這樣也可以實現(xiàn)減少文件數量的目的。這種方法要注意的就是不能無限制的追加，當判斷一個文件已經達到某一個閾值時，就要產生一個新的文件進行追加了。

上述就是小編為大家分享的SparkStreaming如何解決小文件問題了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業(yè)資訊頻道。