package org.dataalgorithms.chap11.projection.secondarysort;

import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Partitioner;
import edu.umd.cloud9.io.pair.PairOfLongInt;

/**
 * NaturalKeyPartitioner
 * 
 * This custom partitioner allow us to distribute how outputs from the 
 * map stage are sent to the reducers.  NaturalKeyPartitioner partitions 
 * the data output from the map phase (SecondarySortProjectionMapper)
 * before it is sent through the shuffle phase. Since we want a single
 * reducer to recieve all projected data for a single "customerID", we 
 * partition data output of the map phase by only the natural key component 
 * ("customerID"). Note that (CompositeKey, PairOfLongInt) is the (key, value)
 * generated by mappers.
 * 
 * 
 * @author Mahmoud Parsian
 *
 */
public class NaturalKeyPartitioner implements
   Partitioner<CompositeKey, PairOfLongInt> {

	@Override
	public int getPartition(CompositeKey key, 
	                        PairOfLongInt value,
			                int numberOfPartitions) {
		return (int) (hash(key.getCustomerID()) % numberOfPartitions);
	}

	@Override
	public void configure(JobConf jobconf) {
	}
	
    /**
     *  adapted from String.hashCode()
     */
    static long hash(String str) {
		return Math.abs(str.hashCode());
    }	

    /**
     *  adapted from String.hashCode()
     */
    static long hash2(String str) {
       long h = 1125899906842597L; // prime
       int length = str.length();
       for (int i = 0; i < length; i++) {
          h = 31*h + str.charAt(i);
       }
       return h;
    }	
}
