hadoop spark合並小文件

 

一.輸入文件類型設置為 CombineTextInputFormat

hadoop

job.setInputFormatClass(CombineTextInputFormat.class)

spark

 val data = sc.newAPIHadoopFile(args(1),
      classOf[CombineTextInputFormat],
      classOf[LongWritable],
      classOf[Text], hadoopConf)
      .map { //TODO }

（hadoop2.7及其以上版本有這個類，雖然2.6也可能用這個類，但不兼容，會出一些bug導致任務失敗；或者直接就報錯找不到類）

二.再配置以下參數：

（如果設置了CombineTextInputFormat而不配置分片大小的參數，所有輸入會合並為一個文件，也就是說，不管你數據多大，只有一個Map，血淚的教訓啊！）

1.運行時加參數

-D mapreduce.input.fileinputformat.split.minsize=134217728
-D mapreduce.input.fileinputformat.split.maxsize=512000000
-D mapred.linerecordreader.maxlength=32768　

例如： hadoop jar xx.jar -D mapreduce.input.fileinputformat.split.minsize=134217728  -D mapreduce.input.fileinputformat.split.maxsize=512000000 <input> <output> 

運行時添加參數這種方法需要在Diver 的main方法第一行添加如下代碼（很重要）：

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

不然直接拿會把運行參數-D當成 args[0],用GenericOptionsParser解析后otherArgs[0]參數是才是<input>;

不習慣運行時添加參數可以直接在Diver類中寫死，代碼中的設置會覆蓋運行時添加的參數。

2.代碼中設置參數

 var hadoopConf = new Configuration()
 hadoopConf.set("mapreduce.input.fileinputformat.split.maxsize", "512000000")
 hadoopConf.set("mapreduce.input.fileinputformat.split.minsize", "268435456")
 hadoopConf.set("mapreduce.input.fileinputformat.split.minsize.per.node", "134217728")   //下面這兩參數可以不設置，詳情看文章末尾
 hadoopConf.set("mapreduce.input.fileinputformat.split.minsize.per.rack", "268435456")

 maxsize和minsize是設置分片上下限的。

（這里還個參數，一般用不上  -D mapred.linerecordreader.maxlength=32768）設置每行最大長度。  

MapReduce中獲取job實例的時候把hadoopConf傳入

Job job = Job.getInstance(hadoopConf,"MyJob");

Spark

//還是上面的spark示例代碼 創建RDD的時候傳入
val data = sc.newAPIHadoopFile(args(1),
      classOf[CombineTextInputFormat],
      classOf[LongWritable],
      classOf[Text], hadoopConf)
      .map { //TODO }

 完畢，打包運行代碼吧！

other：

hadoopconf 的其他兩個參數可以不設置，了解一下

• 如果指定了mapreduce.input.fileinputformat.split.maxsize，那么在同一個節點上的Blocks合並，一個超過maxsize就生成新分片。
• mapreduce.input.fileinputformat.split.minsize.per.node,那么會把1.中處理剩余的Block，進行合並，如果超過minsize，那么全部作為一個分片。否則這些Block與同一機架 Rack上的塊進行合並。
• 每個節點上如上同樣的方式處理，然后針對整個Rack的所有Block，按照1.方式處理。剩余部分，如果指定了mapreduce.input.fileinputformat.split.minsize.per.rack，並且超過minsize.per.rack，則全部作為一個分片，否則這些Block保留，等待與所有機架上的剩余Block進行匯總處理。

每個機架上都按照1，2，3方式處理，匯總所有處理剩下的部分，再按照1的邏輯處理。再剩余的，作為一個分片。