hadoop spark合并小文件

2980315

　　一.输入文件类型设置为 CombineTextInputFormat
　　hadoop
　　

job.setInputFormatClass(CombineTextInputFormat.class)　　

　　spark
　　

val data = sc.newAPIHadoopFile(args(1),
　　>　　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],添加后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的逻辑处理。再剩余的，作为一个分片。