hadoop spark合并小文件
一.输入文件类型设置为 CombineTextInputFormathadoop
job.setInputFormatClass(CombineTextInputFormat.class)
spark
val data = sc.newAPIHadoopFile(args(1),
> classOf,
classOf, 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,添加后otherArgs参数是才是<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,
classOf,
classOf, 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的逻辑处理。再剩余的，作为一个分片。
页:
[1]