目前平臺使用Kafka + Flume的方式進行實時資料接入，Kafka中的資料由業務方負責寫入，這些資料一部分由Spark Streaming進行流式計算；另一部分資料則經由Flume儲存至HDFS，用於資料探勘或機器學習。HDFS儲存資料時目錄的最小邏輯單位為“小時”，為了保證資料計算過程中的資料完整性（計算某個小時目錄中的資料時，該目錄的資料全部寫入完畢，且不再變化），我們在Flume中加入瞭如下策略： 每五分鐘關閉一次正在寫入的檔案，即新建立檔案進行資料寫入。 這樣的方式可以保證，當前小時的第五分鐘之後就可以開始計算上一小時目錄中的資料，一定程度上提高了離線資料處理的實時性。 隨著業務的增加，開始有業務方反饋：“HDFS中實際被分析的資料量很小，但是Spark App的Task數目卻相當多，不太正常”，我們跟進之後，發現問題的根源在於以下三個方面： （1）Kafka的實時資料寫入量比較小； （2）Flume部署多個例項，同時消費Kafka中的資料並寫入HDFS； （3）Flume每五分鐘會重新建立檔案寫入資料（如上所述）； 這樣的場景直接導致HDFS中儲存著數目眾多但單個檔案資料量很小的情況，間接影響著Spark App Task的數目。 我們以Spark WordCount為例進行說明，Spark版本為1.5.1。 假設HDFS目錄“/user/yurun/spark/textfile”中存在以下檔案： 這個目錄下僅三個檔案包含少量資料：part-00005、part-00010、part-00015，資料大小均為6 Byte，其餘檔案資料大小均為0 Byte，符合小檔案的場景。 注意：_SUCCESS相當於一個“隱藏”檔案，實際處理時通常會被忽略。 常規實現 我們使用SparkContext textFile完成資料輸入，應用執行完成之後，通過Spark History Server的頁面可以看到：應用執行過程中，會產生一個Job，包含兩個Stage，每個Stage包含16個Task，也就是說，Task的總數目為32，如下圖所示： 之所以每個Stage包含16個Task，是因為目錄中存有16個文字檔案（_SUCCESS不參與計算）。 優化實現 在這個優化的版本中，我們使用SparkContext newAPIHadoopFile完成資料輸入，需要著重說明一下“org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat”，這個類可以將多個小檔案合併生成一個Split，而一個Split會被一個Task處理，從而減少Task的數目。這個應用的執行過程中，會產生兩個Job，其中Job0包含一個Stage，一個Task；Job1包含兩個Stage，每個Stage包含一個Task，也就是說，Task的總數目為3，如下圖所示：   可以看出，通過使用“org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat”可以很大程度上緩解小檔案導致Spark App Task數目過多的問題。