/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package mapreduce;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 *
 * @author sc-md
 */
public class CustomKey
{          
    public static abstract class Key<KeyValueType>
            implements WritableComparable<Key<KeyValueType>>
    {      
        private Integer sortNumber;
        
        public Key(Integer sequenceNumber)
        {
            super();
            this.sortNumber = sequenceNumber;
        }
               
        public Key setSortNumber(int value)
        {
            sortNumber = value;
            return this;
        }

        public Integer getSortNumber()
        {
            return sortNumber;
        }

        @Override
        public void write(DataOutput out) throws IOException
        {
            out.writeInt(sortNumber);
            writeValue(out);
        }

        @Override
        public void readFields(DataInput in) throws IOException
        {
            sortNumber = in.readInt();
            readValue(in);
        }

        @Override
        public int hashCode()
        {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((getValue() == null) ? 0 : getValue().hashCode());
            result = prime * result + ((sortNumber == null) ? 0 : sortNumber.hashCode());
            return result;
        }

        @Override
        @SuppressWarnings("unchecked")
        public boolean equals(Object obj)
        {
            if (this == obj)
            {
                return true;
            }

            if (obj == null)
            {
                return false;
            }

            if (getClass() != obj.getClass())
            {
                return false;
            }

            return areEqual((Key<KeyValueType>) obj);
        }

        private boolean areEqual(Key<KeyValueType> other)
        {
            if (getValue() == null)
            {
                if (other.getValue() != null)
                {
                    return false;
                }
            }
            else if (!valueEqualTo(other.getValue()))
            {
                return false;
            }

            if (sortNumber == null)
            {
                if (other.sortNumber != null)
                {
                    return false;
                }
            }
            else if (!sortNumber.equals(other.sortNumber))
            {
                return false;
            }
            return true;
        }

        @Override
        public int compareTo(Key<KeyValueType> o)
        {
            // Sort by value to put related data together and
            // then by sequence number so that the data
            // with the lowest sequence number comes first.
            int returnValue = valueCompareTo(o.getValue());
            if (returnValue != 0)
            {
                return returnValue;
            }
            return compareSortOrderNumber(o);
        }

        public int compareSortOrderNumber(Key<KeyValueType> o)
        {
            return Integer.compare(sortNumber, o.sortNumber);
            //return (sortNumber < o.sortNumber ? -1 : (sortNumber == o.sortNumber ? 0 : 1));
        }

        @Override
        public String toString()
        {
            return "CustomKey [key=" + getStringValue() + ", sortNumber=" + sortNumber + "]";
        }
        
        public abstract Key setValue(KeyValueType value);
        public abstract KeyValueType getValue();
        protected abstract String getStringValue();
        protected abstract boolean valueEqualTo(KeyValueType value);
        protected abstract int valueCompareTo(KeyValueType value);
        protected abstract void writeValue(DataOutput out) throws IOException;
        protected abstract void readValue(DataInput in) throws IOException;
    }
    
    //http://www.javamakeuse.com/2016/03/mapreduce-reduce-side-join-example-in.html
    
    // Shuffle (i.e. partition) based on the key value
    public static abstract class KeyPartitioner<KeyType extends Key, KeyValueType> extends Partitioner<KeyType, KeyValueType>
    {
        @Override
        public int getPartition(KeyType key, KeyValueType value, int numPartitions)
        {
            // Divide the keys across the partitions such that the same data
            // for the join key value appears in the same partition. Ignore
            // the sequence number.
            return (key.getValue().hashCode() & Integer.MAX_VALUE) % numPartitions;
        }
    }
    
    // Combine the values from both mappers where they have the same key.
    // This produces the "values" fed to the reduce() function for a specific
    // key.
    public static abstract class KeyGroupComparator<KeyValueType> extends WritableComparator
    {
        protected KeyGroupComparator(Class<? extends WritableComparable> keyClass) {
	  super(keyClass, true);
	 }

         // The "values" associated with each key fed to the Reducer, are determined
        // by the GroupComparator. Here, we want all the values associated with the
        // joinKey value to be grouped together regardless of sequence number.
        @Override
        @SuppressWarnings("unchecked")
        public int compare(WritableComparable w1, WritableComparable w2)
        {
            // Effectively we lose the sort order number because only
            // one key will be passed to the Reducer based on the joinKey value.
            // i.e. only the first key matching the joinKey value will be passed.
            return compareKeyValues(((Key<KeyValueType>)w1).getValue(), ((Key<KeyValueType>)w2).getValue());
        }
        
        public abstract int compareKeyValues(KeyValueType v1, KeyValueType v2);
    }
    
    // Sort the data in the Partition/Group based on the key (i.e. the CustomKey)
    // NOTE: we don't actually need to use this because the compareTo() method
    // in the CustomKey class is used by the default sorter anyway.
//    public static abstract class KeySorter<KeyValueType> extends WritableComparator
//    {
//	 protected KeySorter(Class<? extends WritableComparable> keyClass) {
//	  super(keyClass, true);
//	 }
//
//         // Sort based on the key value and the sequence number
//        @Override
//        public int compare(WritableComparable w1, WritableComparable w2)
//        {
//            Key<KeyValueType> k1 = (Key<KeyValueType>)w1;
//            Key<KeyValueType> k2 = (Key<KeyValueType>)w2;
//            int returnValue = compareKeyValues(k1.getValue(), k2.getValue());
//            if (returnValue != 0)
//            {
//                return returnValue;
//            }
//            return k1.compareSortOrderNumber(k2);
//        }
//        
//        public abstract int compareKeyValues(KeyValueType v1, KeyValueType v2);
//    }
}