Hadoop自定义分区程序问题(Hadoop Custom Partitioner Issue)

 根据自定义分区程序的“getPartition”方法的输出，我遇到的问题是自定义中间键不会出现在我希望的分区中。 我可以在我的映射器日志文件中看到分区器生成预期的分区号，但是有时具有公共分区号的键不会在同一个reducer中结束。 
 
 具有共同“getPartition”输出的键如何以不同的reducers结束？ 
 
 我注意到在所有“getPartition”调用完成后，在映射器日志文件中，对自定义中间键“hashCode”和“compareTo”方法进行了多次调用。 映射器只是在分区排序中进行还是这可能是问题的一部分？ 
 
 我附加了自定义中介密钥和分区程序的代码。 注意：我知道确切的1/2键将“useBothGUIDFlag”设置为true，而1/2将此设置为false（这就是为什么我将这些键分区为分隔空间的不同部分）。 我也知道密钥似乎没有交叉到分区的另一半​​（即“useBothGUIDFlag”键不会在“！useBothGUIDFlag”分区中结束，反之亦然），而是它们在其一半内混合分区。 
 
public class IntermediaryKey implements WritableComparable<IntermediaryKey> {

    public String guid1;
    public String guid2;
    public boolean useBothGUIDFlag;

    @Override
    public int compareTo(IntermediaryKey other) {
        if(useBothGUIDFlag)
        {
            if(other.useBothGUIDFlag)
            {
                return this.hashCode() - other.hashCode();
            }else{
                return 1;
            }
        }else{
            if(!other.useBothGUIDFlag)
            {
                return guid2.compareTo(other.guid2);
            }else{
                return -1;
            }
        }
    }

    @Override
    public int hashCode()
    {
        if(useBothGUIDFlag)
        {
            if(guid1.compareTo(guid2) > 0)
            {
                return (guid2+guid1).hashCode();
            }else{
                return (guid1+guid2).hashCode();
            }
        }else{
            return guid2.hashCode();
        }
    }

    @Override
    public boolean equals(Object otherKey)
    {
        if(otherKey instanceof IntermediaryKey)
        {
            return this.compareTo((IntermediaryKey)otherKey) == 0;
        }
        return false;
    }
}

public static class KeyPartitioner extends Partitioner<IntermediaryKey, PathValue>
{
    @Override
    public int getPartition(IntermediaryKey key, PathValue value, int numReduceTasks) {
        int bothGUIDReducers = numReduceTasks/2;
        if(bothGUIDReducers == 0)
        {
            return 0;
        }

        int keyHashCode = Math.abs(key.hashCode());
        if(key.useBothGUIDFlag)
        {
            return keyHashCode % bothGUIDReducers;
        }else{
            return (bothGUIDReducers + (keyHashCode % (numReduceTasks-bothGUIDReducers)));
        }
    }
}

I am having an issue with a custom intermediary keys not ending up in the partition I would expect based on the output of the custom partitioner's "getPartition" method. I can see in my mapper log files that the partitioner produces the expected partition numbers, however sometimes keys with a common partition number do not end up at the same reducer.
 
How would keys with a common "getPartition" output end up at different reducers?
 
I noticed in the mapper log files after all "getPartition" calls have been made many calls to the custom intermediary keys "hashCode" and "compareTo" methods are made. Is the mapper just doing within partition sorting or could this be part of the issue?
 
I have attached code for the custom intermediary key and partitioner. Note: I know exactly 1/2 of the keys have the "useBothGUIDFlag" set to true and 1/2 have this set to false (which is why I partition these keys to separate halves of the partition space). I also know that keys do not seem to cross over into the other half of the partition (i.e., "useBothGUIDFlag" keys do not end up in the "!useBothGUIDFlag" partitions and vice versa), rather they are mixed up within their half of the partitions.
 
public class IntermediaryKey implements WritableComparable<IntermediaryKey> {

    public String guid1;
    public String guid2;
    public boolean useBothGUIDFlag;

    @Override
    public int compareTo(IntermediaryKey other) {
        if(useBothGUIDFlag)
        {
            if(other.useBothGUIDFlag)
            {
                return this.hashCode() - other.hashCode();
            }else{
                return 1;
            }
        }else{
            if(!other.useBothGUIDFlag)
            {
                return guid2.compareTo(other.guid2);
            }else{
                return -1;
            }
        }
    }

    @Override
    public int hashCode()
    {
        if(useBothGUIDFlag)
        {
            if(guid1.compareTo(guid2) > 0)
            {
                return (guid2+guid1).hashCode();
            }else{
                return (guid1+guid2).hashCode();
            }
        }else{
            return guid2.hashCode();
        }
    }

    @Override
    public boolean equals(Object otherKey)
    {
        if(otherKey instanceof IntermediaryKey)
        {
            return this.compareTo((IntermediaryKey)otherKey) == 0;
        }
        return false;
    }
}

public static class KeyPartitioner extends Partitioner<IntermediaryKey, PathValue>
{
    @Override
    public int getPartition(IntermediaryKey key, PathValue value, int numReduceTasks) {
        int bothGUIDReducers = numReduceTasks/2;
        if(bothGUIDReducers == 0)
        {
            return 0;
        }

        int keyHashCode = Math.abs(key.hashCode());
        if(key.useBothGUIDFlag)
        {
            return keyHashCode % bothGUIDReducers;
        }else{
            return (bothGUIDReducers + (keyHashCode % (numReduceTasks-bothGUIDReducers)));
        }
    }
}

原文：https://stackoverflow.com/questions/13076331