关于Hadoop中的采样器

2216609207

　　1.为什么要使用采样器
　　在这个网页上有一段描述比较靠谱 http://www.philippeadjiman.com/blog/2009/12/20/hadoop-tutorial-series-issue-2-getting-started-with-customized-partitioning/
　　 简单的来说就是解决"How to automatically find “good” partitioning function",因为很多时候无法直接制订固定的partitioner策略,所以需要知道实际的数据分布.糟糕的策略导致的结果就是每个reduce节点得到的数据部均匀,对效率影响挺大
　　
　　2.如何使用采样器
　　


  conf.setPartitionerClass(TotalOrderPartitioner.class);//关于partitioner可以参考这个实现 使用采样器产生的文件

  InputSampler.RandomSampler sampler =
   new InputSampler.RandomSampler(0.1,10000,10);
  
  Path partitionFile = new Path(input,”_partitions”);
  TotalOrderPartitioner.setPartitionFile(conf, partitionFile);////
  InputSampler.writePartitionFile(conf, sampler);
  
//一般都将该文件做distribute cache处理
  URI partitionURI = new URI(partitionFile.toString() + “#_partitions”);
  DistributedCache.addCacheFile(partitionURI, conf);
  DistributedCache.createSymlink(conf);

//从上面可以看出 采样器是在map阶段之前进行的 在提交job的client端完成的
　　3.常用的采样器介绍
　　http://blog.iyunv.com/andyelvis/article/details/7294811
　　Hadoop中采样是由org.apache.hadoop.mapred.lib.InputSampler类来实现的。
　　
InputSampler类实现了三种采样方法：RandomSampler,SplitSampler和IntervalSampler。//RandomSampler最耗时
　　RandomSamplerSplitSampler、RandomSampler和IntervalSampler都是InputSampler的静态内部类，它们都实现了InputSampler的内部接口Sampler接口


public interface Sampler{
      K[] getSample(InputFormat inf,JobConf job) throws IOException;
}　　getSample方法根据job的配置信息以及输入格式获得抽样结果，三个采样类各自有不同的实现。
　　
RandomSampler随机地从输入数据中抽取Key，是一个通用的采样器。RandomSampler类有三个属性：freq（一个Key被选中的概率），numSamples（从所有被选中的分区中获得的总共的样本数目），maxSplitsSampled（需要检查扫描的最大分区数目）。
RandomSampler中getSample方法的实现如下：


    public K[] getSample(InputFormat inf, JobConf job) throws IOException {
      InputSplit[] splits = inf.getSplits(job, job.getNumMapTasks());
      ArrayList samples = new ArrayList(numSamples);
      int splitsToSample = Math.min(maxSplitsSampled, splits.length);
      Random r = new Random();
      long seed = r.nextLong();
      r.setSeed(seed);
      LOG.debug("seed: " + seed);
      // shuffle splits
      for (int i = 0; i < splits.length; ++i) {
        InputSplit tmp = splits;
        int j = r.nextInt(splits.length);
        splits = splits[j];
        splits[j] = tmp;
      }
      // our target rate is in terms of the maximum number of sample splits,
      // but we accept the possibility of sampling additional splits to hit
      // the target sample keyset
      for (int i = 0; i < splitsToSample ||
                     (i < splits.length && samples.size() < numSamples); ++i) {
        RecordReader reader = inf.getRecordReader(splits, job,
            Reporter.NULL);
        K key = reader.createKey();
        V value = reader.createValue();
        while (reader.next(key, value)) {
          if (r.nextDouble()