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的逻辑处理。再剩余的,作为一个分片。

上一篇:Hadoop HDFS编程 API入门系列之合并小文件到HDFS(三)


下一篇:安装64位Oracle 10g超详细教程