一、背景
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(InputFormatinf, 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(); }
评论列表()