Hadoop对文本文件的快速全局排序实现方法及分析

一、背景

Hadoop中实现了用于全局排序的InputSampler类和TotalOrderPartitioner类，调用示例是org.apache.hadoop.examples.Sort。

但是当我们以Text文件作为输入时，结果并非按Text中的string列排序，而且输出结果是SequenceFile。

原因：

1） hadoop在处理Text文件时，key是行号LongWritable类型，InputSampler抽样的是key，TotalOrderPartitioner也是用key去查找分区。这样，抽样得到的partition文件是对行号的抽样，结果自然是根据行号来排序。

2）大数据量时，InputSampler抽样速度会非常慢。比如，RandomSampler需要遍历所有数据，IntervalSampler需要遍历文件数与splits数一样。SplitSampler效率比较高，但它只抽取每个文件前面的记录，不适合应用于文件内有序的情况。

二、功能

1. 实现了一种局部抽样方法PartialSampler，适用于输入数据各文件是独立同分布的情况

2. 使RandomSampler、IntervalSampler、SplitSampler支持对文本的抽样

3. 实现了针对Text文件string列的TotalOrderPartitioner

三、实现

1. PartialSampler

PartialSampler从第一份输入数据中随机抽取第一列文本数据。PartialSampler有两个属性：freq（采样频率），numSamples（采样总数）。

public K[] getSample(InputFormat inf, JobConf job) throws IOException {   InputSplit[] splits = inf.getSplits(job, job.getNumMapTasks());   ArrayList samples = new ArrayList(numSamples);   Random r = new Random();   long seed = r.nextLong();   r.setSeed(seed);   LOG.debug("seed: " + seed);      // 对splits【0】抽样   for (int i = 0; i < 1; i++) {    System.out.println("PartialSampler will getSample splits["+i+"]");    RecordReader reader = inf.getRecordReader(splits[i], job,      Reporter.NULL);    K key = reader.createKey();    V value = reader.createValue();    while (reader.next(key, value)) {     if (r.nextDouble() <= freq) {      if (samples.size() < numSamples) {        // 选择value中的第一列抽样        Text value0 = new Text(value.toString().split("/t")[0]);             samples.add((K) value0);              } else {       // When exceeding the maximum number of samples, replace a       // random element with this one, then adjust the frequency       // to reflect the possibility of existing elements being       // pushed out       int ind = r.nextInt(numSamples);       if (ind != numSamples) {        Text value0 = new Text(value.toString().split("/t")[0]);         samples.set(ind, (K) value0);       }       freq *= (numSamples - 1) / (double) numSamples;      }      key = reader.createKey();     }    }        reader.close();   }   return (K[])samples.toArray();  }