Partition所處的位置

Partition主要作用就是將map的結果發送到相應的reduce。這就對partition有兩個要求：

1）均衡負載，盡量的將工作均勻的分配給不同的reduce。

2）效率，分配速度一定要快。

Mapreduce提供的Partitioner

Mapreduce默認的partitioner是HashPartitioner。除了這個mapreduce還提供了3種partitioner。如下圖所示：

2. HashPartitioner是mapreduce的默認partitioner。計算方法是

which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到當前的目的reducer。

3. BinaryPatitioner繼承于Partitioner< BinaryComparable ,V>，是Partitioner的偏特化子類。該類提供leftOffset和rightOffset，在計算which reducer時僅對鍵值K的[rightOffset，leftOffset]這個區間取hash。

Which reducer=(hash & Integer.MAX_VALUE) % numReduceTasks

4. KeyFieldBasedPartitioner也是基于hash的個partitioner。和BinaryPatitioner不同，它提供了多個區間用于計算hash。當區間數為0時KeyFieldBasedPartitioner退化成HashPartitioner。

5. TotalOrderPartitioner這個類可以實現輸出的全排序。不同于以上3個partitioner，這個類并不是基于hash的。在下一節里詳細的介紹totalorderpartitioner。

TotalOrderPartitioner

每一個reducer的輸出在默認的情況下都是有順序的，但是reducer之間在輸入是無序的情況下也是無序的。如果要實現輸出是全排序的那就會用到TotalOrderPartitioner。

要使用TotalOrderPartitioner，得給TotalOrderPartitioner提供一個partition file。這個文件要求Key （這些key就是所謂的劃分）的數量和當前reducer的數量-1相同并且是從小到大排列。對于為什么要用到這樣一個文件，以及這個文件的具體細節待會還會提到。

TotalOrderPartitioner對不同Key的數據類型提供了兩種方案：

1） 對于非BinaryComparable（參考附錄A）類型的Key，TotalOrderPartitioner采用二分發查找當前的K所在的index。

例如reducer的數量為5，partition file 提供的4個劃分為【2，4，6，8】。如果當前的一個key value pair 是<4,”good”>利用二分法查找到index=1，index+1=2那么這個key value pair將會發送到第二個reducer。如果一個key value pair為<4.5, “good”>那么二分法查找將返回-3，同樣對-3加1然后取反就是這個key value pair 將要去的reducer。

對于一些數值型的數據來說，利用二分法查找復雜度是o（log (reducer count)），速度比較快。

2） 對于BinaryComparable類型的Key（也可以直接理解為字符串）。字符串按照字典順序也是可以進行排序的。這樣的話也可以給定一些劃分，讓不同的字符串key分配到不同的reducer里。這里的處理和數值類型的比較相近。

例如reducer的數量為5，partition file 提供了4個劃分為【“abc”, “bce”, “eaa”, ”fhc”】那么“ab”這個字符串將會被分配到第一個reducer里，因為它小于第一個劃分“abc”。

但是不同于數值型的數據，字符串的查找和比較不能按照數值型數據的比較方法。mapreducer采用的Tire tree的字符串查找方法。查找的時間復雜度o(m)，m為樹的深度，空間復雜度o(255^m-1)。是一個典型的空間換時間的案例。

Tire Tree

Tire tree的構建

假設樹的最大深度為3，劃分為【aaad ，aaaf， aaaeh，abbx 】

Tire tree的搜索過程

接上面的例子：
1）假如當前 key value pair這時會找到圖中的leafnode，在leafnode內部使用二分法繼續查找找到返回 aad在 劃分數組中的索引。找不到會返回一個和它最接近的劃分的索引。
2）假如找到singlenode，如果和singlenode的劃分相同或小返回他的索引，比singlenode的劃分大則返回索引+1。
3）假如找到nosplitnode則返回前面的索引。如將會返回abbx的在劃分數組中的索引。

TotalOrderPartitioner的疑問

上面介紹了partitioner有兩個要求，一個是速度另外一個是均衡負載。使用tire tree提高了搜素的速度，但是我們怎么才能找到這樣的partition file 呢？讓所有的劃分剛好就能實現均衡負載。

InputSampler
輸入采樣類，可以對輸入目錄下的數據進行采樣。提供了3種采樣方法。

采樣方式對比表:

writePartitionFile這個方法很關鍵，這個方法就是根據采樣類提供的樣本，首先進行排序，然后選定（隨機的方法）和reducer數目-1的樣本寫入到partition file。這樣經過采樣的數據生成的劃分，在每個劃分區間里的key value pair 就近似相同了，這樣就能完成均衡負載的作用。

TotalOrderPartitioner實例

    

      

        
          
            
          
        
      
      

        
          public
        
        
          class
        
        SortByTemperatureUsingTotalOrderPartitioner 
   
        
          extends
        
        Configured 
   
        
          implements
        
        Tool { @Override 
   
        
          public
        
        
          int
        
        run(String[]args) 
   
        
          throws
        
        Exception { JobConfconf=JobBuilder.parseInputAndOutput( 
   
        
          this
        
        ,getConf(),args); 
   
        
          if
        
        (conf== 
   
        
          null
        
        ){ 
   
        
          return
        
        -1; } conf.setInputFormat(SequenceFileInputFormat. 
   
        
          class
        
        ); conf.setOutputKeyClass(IntWritable. 
   
        
          class
        
        ); conf.setOutputFormat(SequenceFileOutputFormat. 
   
        
          class
        
        ); SequenceFileOutputFormat.setCompressOutput(conf, 
   
        
          true
        
        ); SequenceFileOutputFormat .setOutputCompressorClass(conf,GzipCodec. 
   
        
          class
        
        ); SequenceFileOutputFormat.setOutputCompressionType(conf, CompressionType.BLOCK); conf.setPartitionerClass(TotalOrderPartitioner. 
   
        
          class
        
        ); InputSampler.Sampler<IntWritable,Text>sampler= 
   
        
          new
        
        InputSampler.RandomSampler<IntWritable,Text>( 0.1,10000,10); Pathinput=FileInputFormat.getInputPaths(conf)[0]; input=input.makeQualified(input.getFileSystem(conf)); PathpartitionFile= 
   
        
          new
        
        Path(input,"_partitions"); TotalOrderPartitioner.setPartitionFile(conf,partitionFile); InputSampler.writePartitionFile(conf,sampler); 
   
        
          //
        
        
          AddtoDistributedCache
        
        
        
        URIpartitionUri= 
   
        
          new
        
        URI(partitionFile.toString()+"#_partitions"); DistributedCache.addCacheFile(partitionUri,conf); DistributedCache.createSymlink(conf); JobClient.runJob(conf); 
   
        
          return
        
        0; } 
   
        
          public
        
        
          static
        
        
          void
        
        main(String[]args) 
   
        
          throws
        
        Exception{ 
   
        
          int
        
        exitCode=ToolRunner.run( 
   
        
          new
        
        SortByTemperatureUsingTotalOrderPartitioner(),args); System.exit(exitCode); } } 
  
      
      

        
          
            
          
        
      
    
  

示例程序引用于：http://www.cnblogs.com/funnydavid/archive/2010/11/24/1886974.html

附錄A
Text 為BinaryComparable，WriteableComparable類型。
BooleanWritable、ByteWritable、DoubleWritable、MD5hash、IntWritable、FloatWritable、LongWritable、NullWriable等都為WriteableComparable。

http://www.cnblogs.com/OnlyXP/archive/2008/12/06/1349026.html

Hadoop 的 TotalOrderPartitioner