ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > MapReduce 大量小文件

MapReduce 大量小文件

原创 Hadoop 作者：破棉袄 时间：2014-07-23 10:45:20 0 删除 编辑

我们知道Map的数量一般是由分片数量决定的，而当输入为大量的小文件时，可能就会产生大量的Map任务，并且每个Map任务的任务量都很小。
所以，我们需要将多个小文件交由一个Map 任务来处理，开始想自己写一套InputFormat、RecordReader、FileSplit。后来得知其实Hadoop早已为
我们提供了CombineFileInputFormat和CombineFileInputFormat。 （提供的CombineFileInputFormat会产生数据倾斜问题，后边讲）

CombineFileInputFormat的大致原理是，他会将输入多个数据文件（小文件）的元数据全部包装到CombineFileSplit类里面。也就是说，因为小文件的情况下，在HDFS中都是单Block的文件，即一个文件一个Block，一个CombineFileSplit包含了一组文件Block，包括每个文件的起始偏移（offset），长度（length），Block位置（localtions）等元数据。

（代码大多采用自其它兄弟的博文，自己只是吸收完后在此记录以作备忘，在此先谢过）

1.先看一下CombineFileSplit 的源码：（简简单单五个对象）
  1. public class CombineFileSplit extends InputSplit implements Writable {

  3.   private Path[] paths;
  4.   private long[] startoffset;
  5.   private long[] lengths;
  6.   private String[] locations;
  7.   private long totLength;

  9.   /**
  10.    * default constructor
  11.    */
  12.   public CombineFileSplit() {}
  13.   public CombineFileSplit(Path[] files, long[] start,
  14.                           long[] lengths, String[] locations) {
  15.     initSplit(files, start, lengths, locations);
  16.   }

  18.   public CombineFileSplit(Path[] files, long[] lengths) {
  19.     long[] startoffset = new long[files.length];
  20.     for (int i = 0; i < startoffset.length; i++) {
  21.       startoffset[i] = 0;
  22.     }
  23.     String[] locations = new String[files.length];
  24.     for (int i = 0; i < locations.length; i++) {
  25.       locations[i] = "\";
  26.     }
  27.     initSplit(files, startoffset, lengths, locations);
  28.   }
  29.   
  30.   private void initSplit(Path[] files, long[] start,
  31.                          long[] lengths, String[] locations) {
  32.     this.startoffset = start;
  33.     this.lengths = lengths;
  34.     this.paths = files;
  35.     this.totLength = 0;
  36.     this.locations = locations;
  37.     for(long length : lengths) {
  38.       totLength += length;
  39.     }
  40.   }

  42.   /**
  43.    * Copy constructor
  44.    */
  45.   public CombineFileSplit(CombineFileSplit old) throws IOException {
  46.     this(old.getPaths(), old.getStartOffsets(),
  47.          old.getLengths(), old.getLocations());
  48.   }

  50.   public long getLength() {
  51.     return totLength;
  52.   }

  54.   /** Returns an array containing the start offsets of the files in the split*/
  55.   public long[] getStartOffsets() {
  56.     return startoffset;
  57.   }
  58.   
  59.   /** Returns an array containing the lengths of the files in the split*/
  60.   public long[] getLengths() {
  61.     return lengths;
  62.   }

  64.   /** Returns the start offset of the i<sup>th</sup> Path */
  65.   public long getOffset(int i) {
  66.     return startoffset[i];
  67.   }
  68.   
  69.   /** Returns the length of the i<sup>th</sup> Path */
  70.   public long getLength(int i) {
  71.     return lengths[i];
  72.   }
  73.   
  74.   /** Returns the number of Paths in the split */
  75.   public int getNumPaths() {
  76.     return paths.length;
  77.   }

  79.   /** Returns the i<sup>th</sup> Path */
  80.   public Path getPath(int i) {
  81.     return paths[i];
  82.   }
  83.   
  84.   /** Returns all the Paths in the split */
  85.   public Path[] getPaths() {
  86.     return paths;
  87.   }

  89.   /** Returns all the Paths where this input-split resides */
  90.   public String[] getLocations() throws IOException {
  91.     return locations;
  92.   }

  94.   public void readFields(DataInput in) throws IOException {
  95.     totLength = in.readLong();
  96.     int arrLength = in.readInt();
  97.     lengths = new long[arrLength];
  98.     for(int i=0; i<arrLength;i++) {
  99.       lengths[i] = in.readLong();
  100.     }
  101.     int filesLength = in.readInt();
  102.     paths = new Path[filesLength];
  103.     for(int i=0; i<filesLength;i++) {
  104.       paths[i] = new Path(Text.readString(in));
  105.     }
  106.     arrLength = in.readInt();
  107.     startoffset = new long[arrLength];
  108.     for(int i=0; i<arrLength;i++) {
  109.       startoffset[i] = in.readLong();
  110.     }
  111.   }

  113.   public void write(DataOutput out) throws IOException {
  114.     out.writeLong(totLength);
  115.     out.writeInt(lengths.length);
  116.     for(long length : lengths) {
  117.       out.writeLong(length);
  118.     }
  119.     out.writeInt(paths.length);
  120.     for(Path p : paths) {
  121.       Text.writeString(out, p.toString());
  122.     }
  123.     out.writeInt(startoffset.length);
  124.     for(long length : startoffset) {
  125.       out.writeLong(length);
  126.     }
  127.   }
  128.   
  129.   @Override
  130.  public String toString() {
  131.     StringBuffer sb = new StringBuffer();
  132.     for (int i = 0; i < paths.length; i++) {
  133.       if (i == 0 ) {
  134.         sb.append("Paths:");
  135.       }
  136.       sb.append(paths[i].toUri().getPath() + ":" + startoffset[i] +
  137.                 "+" + lengths[i]);
  138.       if (i < paths.length -1) {
  139.         sb.append(",");
  140.       }
  141.     }
  142.     if (locations != null) {
  143.       String locs = "";
  144.       StringBuffer locsb = new StringBuffer();
  145.       for (int i = 0; i < locations.length; i++) {
  146.         locsb.append(locations[i] + ":");
  147.       }
  148.       locs = locsb.toString();
  149.       sb.append(" Locations:" + locs + "; ");
  150.     }
  151.     return sb.toString();
  152.   }
  153. }

2.接下来我们需要来实现一个RecordReader，这个RecordReader其实为LineRecordReader的封装
  1. import java.io.IOException;

  3. import org.apache.hadoop.fs.Path;
  4. import org.apache.hadoop.io.BytesWritable;
  5. import org.apache.hadoop.io.LongWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.InputSplit;
  8. import org.apache.hadoop.mapreduce.RecordReader;
  9. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  10. import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
  11. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  12. import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

  14. public class CombineSmallfileRecordReader extends RecordReader<LongWritable,Text> {

  16.     private CombineFileSplit combineFileSplit;
  17.     private LineRecordReader lineRecordReader = new LineRecordReader();
  18.     private Path[] paths;
  19.     private int totalLength;
  20.     private int currentIndex;
  21.     private float currentProgress = 0;
  22.     private LongWritable currentKey;
  23.     private Text currentValue = new Text();

  25.     public CombineSmallfileRecordReader(CombineFileSplit combineFileSplit, TaskAttemptContext context, Integer index) throws IOException {
  26.         super();
  27.         this.combineFileSplit = combineFileSplit;
  28.         this.currentIndex = index; // 当前要处理的小文件Block在CombineFileSplit中的索引
  29.     }

  31.     @Override
  32.     public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
  33.         this.combineFileSplit = (CombineFileSplit) split;
  34.         // 处理CombineFileSplit中的一个小文件Block，因为使用LineRecordReader，需要构造一个FileSplit对象，然后才能够读取数据
  35.         FileSplit fileSplit = new FileSplit(combineFileSplit.getPath(currentIndex), combineFileSplit.getOffset(currentIndex), combineFileSplit.getLength(currentIndex), combineFileSplit.getLocations());
  36.         lineRecordReader.initialize(fileSplit, context);

  38.         this.paths = combineFileSplit.getPaths();
  39.         totalLength = paths.length;
  40.         context.getConfiguration().set("map.input.file.name", combineFileSplit.getPath(currentIndex).getName());
  41.     }

  43.     @Override
  44.     public LongWritable getCurrentKey() throws IOException, InterruptedException {
  45.         currentKey = lineRecordReader.getCurrentKey();
  46.         return currentKey;
  47.     }

  49.     @Override
  50.     public Text getCurrentValue() throws IOException, InterruptedException {
  51.         currentValue = lineRecordReader.getCurrentValue();
            return currentValue;
  52.     }

  54.     @Override
  55.     public boolean nextKeyValue() throws IOException, InterruptedException {
  56.         if (currentIndex >= 0 && currentIndex < totalLength) {
  57.             return lineRecordReader.nextKeyValue();
  58.         } else {
  59.             return false;
  60.         }
  61.     }

  63.     @Override
  64.     public float getProgress() throws IOException {
  65.         if (currentIndex >= 0 && currentIndex < totalLength) {
  66.             currentProgress = (float) currentIndex / totalLength;
  67.             return currentProgress;
  68.         }
  69.         return currentProgress;
  70.     }

  72.     @Override
  73.     public void close() throws IOException {
  74.         lineRecordReader.close();
  75.     }
  76. }

3.接着实现一个CombineFileInputFormat的子类
  1. import java.io.IOException;

  3. import org.apache.hadoop.io.BytesWritable;
  4. import org.apache.hadoop.io.LongWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.InputSplit;
  7. import org.apache.hadoop.mapreduce.RecordReader;
  8. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  9. import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
  10. import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
  11. import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;

  13. public class CombineSmallfileInputFormat extends CombineFileInputFormat<LongWritable,Text> {

  15.     @Override
  16.     public RecordReader<LongWritable,Text> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {

  18.         CombineFileSplit combineFileSplit = (CombineFileSplit) split;
  19.         CombineFileRecordReader<LongWritable,Text> recordReader = new CombineFileRecordReader<LongWritable,Text>(combineFileSplit, context, CombineSmallfileRecordReader.class);
  20.         try {
  21.             recordReader.initialize(combineFileSplit, context);
  22.         } catch (InterruptedException e) {
  23.             new RuntimeException("Error to initialize CombineSmallfileRecordReader.");
  24.         }
  25.         return recordReader;
  26.     }

  28. }

4.至此我们已经完成了小文件输入的问题，但是让我们看一下 CombineInputFormat 的分片逻辑

         CombineInputFormat 中利用三个参数 maxSplitSize、minSizeNode、minSizeRack 来控制分片：
        1.如果指定了maxSplitSize(“mapreduce.input.fileinputformat.split.maxsize”)，那么在同一个节点上的Blocks合并，一个超过maxSplitSize就生成新分片。如果没有指定，则只汇总本节点BLock，暂不分片。
         2.如果指定了minSizeNode(“mapreduce.input.fileinputformat.split.minsize.per.node”),那么会把1.中处理剩余的Block，进行合并，如果超过minSizeNode，那么全部作为一个分片。否则这些Block与同一机架     Rack上的块进行合并。
        3.每个节点上如上同样的方式处理，然后针对整个Rack的所有Block，按照1.方式处理。剩余部分，如果指定了minSizeRack(“mapreduce.input.fileinputformat.split.minsize.per.rack”)，并且超过minSizeRack，则全部作为一个分片，否则这些Block保留，等待与所有机架上的剩余Block进行汇总处理。

      每个机架上都按照1，2，3方式处理，汇总所有处理剩下的部分，再按照1的逻辑处理。再剩余的，作为一个分片。

以上逻辑我们可以知道：

如果只设置maxSplitSize(如job.getConfiguration().set( “mapreduce.input.fileinputformat.split.maxsize” , “33554432″))，那么基本每个分片大小都需凑满maxSplitSize。

如果maxSplitSize，minSizeNode，minSizeRack三个都没有设置，那是所有输入整合成一个分片！

CombineInputFormat 提供的分片机制实在是复杂，并且可能会出现数据倾斜问题。所以我们来自己Override一个简单实用的getSplits方法：
  1. private static final Log LOG = LogFactory.getLog(MultiFileInputFormat.class);
  2.     public static final String CONFNAME_INPUT_SPLIT_MAX_NUM = "multifileinputformat.max_split_num";
  3.     public static final Integer DEFAULT_MAX_SPLIT_NUM = 50;

  5.     public static void setMaxInputSplitNum(Job job, Integer maxSplitNum) {
  6.         job.getConfiguration().setInt(CONFNAME_INPUT_SPLIT_MAX_NUM, maxSplitNum);
  7.     }

  9.     @Override
  10.     public List<InputSplit> getSplits(JobContext job) throws IOException {
  11.         // get all the files in input path
  12.         List<FileStatus> stats = listStatus(job);
  13.         List<InputSplit> splits = new ArrayList<InputSplit>();
  14.         if (stats.size() == 0) {
  15.             return splits;
  16.         }
  17.         // 计算split的平均长度
  18.         long totalLen = 0;
  19.         for (FileStatus stat : stats) {
  20.             totalLen += stat.getLen();
  21.         }
  22.         int maxSplitNum = job.getConfiguration().getInt(CONFNAME_INPUT_SPLIT_MAX_NUM, DEFAULT_MAX_SPLIT_NUM);
  23.         int expectSplitNum = maxSplitNum < stats.size() ? maxSplitNum : stats.size();
  24.         long averageLen = totalLen / expectSplitNum;
  25.         LOG.info("Prepare InputSplit : averageLen(" + averageLen + ") totalLen(" + totalLen
  26.                 + ") expectSplitNum(" + expectSplitNum + ") ");
  27.         // 设置inputSplit
  28.         List<Path> pathLst = new ArrayList<Path>();
  29.         List<Long> offsetLst = new ArrayList<Long>();
  30.         List<Long> lengthLst = new ArrayList<Long>();
  31.         long currentLen = 0;
  32.         for (int i = 0; i < stats.size(); i++) {
  33.             FileStatus stat = stats.get(i);
  34.             pathLst.add(stat.getPath());
  35.             offsetLst.add(0L);
  36.             lengthLst.add(stat.getLen());
  37.             currentLen += stat.getLen();
  38.             if (splits.size() < expectSplitNum - 1 && currentLen > averageLen) {
  39.                 Path[] pathArray = new Path[pathLst.size()];
  40.                 CombineFileSplit thissplit = new CombineFileSplit(pathLst.toArray(pathArray),
  41.                     getLongArray(offsetLst), getLongArray(lengthLst), new String[0]);
  42.                 LOG.info("combineFileSplit(" + splits.size() + ") fileNum(" + pathLst.size()
  43.                         + ") length(" + currentLen + ")");
  44.                 splits.add(thissplit);
  45.                 
  46.                 pathLst.clear();
  47.                 offsetLst.clear();
  48.                 lengthLst.clear();
  49.                 currentLen = 0;
  50.             }
  51.         }
  52.         if (pathLst.size() > 0) {
  53.             Path[] pathArray = new Path[pathLst.size()];
  54.             CombineFileSplit thissplit =
  55.                     new CombineFileSplit(pathLst.toArray(pathArray), getLongArray(offsetLst),
  56.                             getLongArray(lengthLst), new String[0]);
  57.             LOG.info("combineFileSplit(" + splits.size() + ") fileNum(" + pathLst.size()
  58.                     + ") length(" + currentLen + ")");
  59.             splits.add(thissplit);
  60.         }
  61.         return splits;
  62.     }

  64.     private long[] getLongArray(List<Long> lst) {
  65.         long[] rst = new long[lst.size()];
  66.         for (int i = 0; i < lst.size(); i++) {
  67.             rst[i] = lst.get(i);
  68.         }
  69.         return rst;
  70.     }
这样就可以通过 multifileinputformat.max_split_num 参数准确的控制map数量，且会均匀的分发数据到各Map

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/29754888/viewspace-1225105/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 巅峰对决：Hypertable(C++)吞吐率测试完胜HBase(Java)
下一篇： Hadoop的fsck工具
请登录后发表评论 登录
全部评论
破棉袄

注册时间：2014-07-16

  • 博文量
    180
  • 访问量
    1158161

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593