代码测试环境:Hadoop2.4
应用场景:当需要处理非常多的小数据文件,这种技术的目的,可以被应用到实现高效的数据处理。
原理:申请书CombineFileInputFormat,能够进行切片合并的时候把多个小的数据文件。因为每个切片将有一个Mapper,当一个Mapper处理的数据比較小的时候,其效率较低。而一般使用Hadoop处理数据时。即默认方式,会把一个输入数据文件当做一个分片。这样当输入文件较小时就会出现效率低下的情况。
实例:
參考前篇blog:。只是这次输入使用两个输入文件,都是小数据量的数据文件。
自己定义输入文件格式:CustomCombineFileInputFormat:
package fz.combineinputformat;import java.io.IOException;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.InputSplit;import org.apache.hadoop.mapreduce.RecordReader;import org.apache.hadoop.mapreduce.TaskAttemptContext;import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;/** * 定义读取类 * @author fansy * */public class CustomCombineFileInputFormat extends CombineFileInputFormat自己定义记录读取类CustomCombineReader:{ @Override public RecordReader createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException { // TODO Auto-generated method stub return new CombineFileRecordReader ((CombineFileSplit)split,context,CustomCombineReader.class); }}
package fz.combineinputformat;import java.io.IOException;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.InputSplit;import org.apache.hadoop.mapreduce.RecordReader;import org.apache.hadoop.mapreduce.TaskAttemptContext;import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;import org.apache.hadoop.mapreduce.lib.input.FileSplit;/** * 改动初始化函数 * @author fansy * */public class CustomCombineReader extends RecordReader能够看到这个类使用了上篇博客的CustomReader类。仅仅是改动了下初始化函数,使得小数据量的文件能够合并到一个分片而已。CustomReader能够參考前篇blog: 。{ private int index; private CustomReader in; public CustomCombineReader(CombineFileSplit split,TaskAttemptContext cxt,Integer index){ this.index=index; this.in= new CustomReader(); } @Override public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { CombineFileSplit cfsplit= (CombineFileSplit) split; FileSplit fileSplit = new FileSplit(cfsplit.getPath(index),cfsplit.getOffset(index), cfsplit.getLength(),cfsplit.getLocations()); in.initialize(fileSplit, context); } @Override public boolean nextKeyValue() throws IOException, InterruptedException { return in.nextKeyValue(); } @Override public Text getCurrentKey() throws IOException, InterruptedException { // TODO Auto-generated method stub return in.getCurrentKey(); } @Override public Text getCurrentValue() throws IOException, InterruptedException { // TODO Auto-generated method stub return in.getCurrentValue(); } @Override public float getProgress() throws IOException, InterruptedException { // TODO Auto-generated method stub return in.getProgress(); } @Override public void close() throws IOException { // TODO Auto-generated method stub in.close(); }}
主类,仅仅需改动(相同參考前篇blog):
job.setInputFormatClass(CustomCombineFileInputFormat.class);进行了两次实验。第一次使用CombineFileInputFormat读取,第二次使用TextInputFormat读取。
结果查看:
首先能够从终端看出来:
能够看到相同的两个输入文件,任务096仅仅有一个分片。任务097有两个分片;
同一时候在任务监控界面也能够看到Mapper的个数变化:
总结:CombineFileInputFormat具有非常强的应用价值,针对大量小数据具有非常高的处理效率收益。只是。假设是大数据应用,普通情况下可能输入数据都是非常大的,所以。这样的情况也仅仅是针对一些特殊情况的处理。
分享,成长。快乐
转载请注明blog地址:
版权声明:本文博主原创文章,博客,未经同意不得转载。