Home | About | Sematext search-lucene.com search-hadoop.com
 Search Hadoop and all its subprojects:

Switch to Threaded View
HDFS, mail # user - RE: Why the reducer's input group count is higher than my GroupComparator implementation


Copy link to this message
-
RE: Why the reducer's input group count is higher than my GroupComparator implementation
java8964 java8964 2013-10-31, 01:30
OK.
Finally I found out the reason for this problem, after reading the Hadoop ReducerTask source code.
Here is the mistake I made, and something new I learned about secondary sorting.
In my originally implementation, I partitioned the data by YYYYMMDD, but I didn't add this information in the sort order. That said, if you want to group and partition the data based on field1, field2, and sort the data on field 3, field, you have to sort the data based on field1, field2 first, then sort remaining based on what order you want. My mistake is that I didn't sort on field 2, even I do group and partition also on field 2.
New thing I learned after reading source code is that the NOT just one group data could send to one reducer, but multi groups of data. I think this makes sense, as you could have a lot of input groups data, but only small number of reducers to handle them. But the trick is that it is both Grouping Comparator and the sort order together decide if a new input group of reducer created or not.
For example, in my data, if I have:
type1, id1, YYYYMMDD1type1, id2, YYYYMMDD2type1, id3, YYYYMMDD1type1, id4, YYYYMMDD2
These will consider as 4 input group to the reducer, instead of 2 (Even the data only have (type1, YYYYMMDD1) and (type1, YYYYMMDD2) 2 unique values), but since YYYYMMDD is not part of the sorting order, so the data could arrive in the above order, and the GroupCompare will get non-zero result for each compare, then leads to 4 input groups generated.
Lesson learned, and be able to read the source code is a huge benefit.
Yong
From: [EMAIL PROTECTED]
To: [EMAIL PROTECTED]
Subject: RE: Why the reducer's input group count is higher than my GroupComparator implementation
Date: Tue, 29 Oct 2013 09:46:34 -0400
Yes.
The Partitioner uses the same hashcode() on the String generated from the (type + YYYY/MM/DD).
I add the log in the GroupComparator, and observed there are only 11 unique values being compared in the GroupComparator, but don't know why the reducers input group number is much higher than 11.

Date: Tue, 29 Oct 2013 08:57:32 +0100
Subject: Re: Why the reducer's input group count is higher than my GroupComparator implementation
From: [EMAIL PROTECTED]
To: [EMAIL PROTECTED]

Did you overwrite the partitioner as well?
2013/10/29 java8964 java8964 <[EMAIL PROTECTED]>
Hi, I have a strange question related to my secondary sort implementation in the MR job.Currently I need to support 2nd sort in one of my MR job. I implemented my custom WritableComparable like following:

public class MyPartitionKey implements WritableComparable<MyPartitionKey> {    String type;    long id1;    String id2;    String id3;    String id4;
    long timestamp1;    long timestamp2}
Then I implemented following methods for this class:
public int compareTo(); // sort the data based on all attributes listed above, sorted the last 2 timestamps descending
public int hashCode(); // generate the hashcode using all attributes abovepublic boolean equals(); // using all the attributes for equals checkpublic void write(DataOutput out) // serialize all the attributes listed above
public void readFields(DataInput in) // deserialize all the attributes listed above
For partition and grouping of my keys, I want the following logic:Based on the type, the data could partition either by year or by day for timestamp1.

For sorting order, I want the data sort by (type, id1, id2, id3, id4), then reverse sorting by (timestamp1, timestamp2).
I implemented my KeyComparator using my sorting order logic listed above, and my Partitioner and GroupComparator based on my logic listed above.

Here is the pseudo code of the Partitioner and GroupComparator:
public class MyPartitioner implements Partitioner {    @Override    public int getPartition(MyPartitionKey key, Value value, int numPartitions) {
        int hashCode = key.getActivityType().name().hashCode();        StringBuilder sb = new StringBuilder();        for (String subPartitionValue : key.getPartitionValue()) {            sb.append(subPartitionValue);
        }        return Math.abs(hashCode * 127 + sb.toString().hashCode()) % numPartitions;    }
    @Override    public void configure(JobConf job) {
    }}
// The key getPartitionValue method will return array of string of either YYYY or {YYYY, MM, DD} of the timestamp1.
For GroupComparator:

    public static class MyGroupComparator extends WritableComparator {        protected MyGroupComparator() {            super(MyPartitionKey.class, true);
        }
        @Override        public int compare(WritableComparable w1, WritableComparable w2) {            MyPartitionKey key1 = (MyPartitionKey) w1;            MyPartitionKey key2 = (MyPartitionKey) w2;
            int cmp = key1.type.compareTo(key2.type);            // different type, send to different group            if (cmp != 0)                return cmp;

            // for the same type, should have the same partition value array length            String[] partitionValue1 = key1.getPartitionValue();
            String[] partitionValue2 = key2.getPartitionValue();            assert partitionValue1.length == partitionValue2.length;            StringBuilder sb1 = new StringBuilder();            StringBuilder sb2 = new StringBuilder();
            for (String subValue : partitionValue1) {                sb1.append(subValue);            }            for (String subValue : partitionValue2) {                sb2.append(subValue);
            }            return sb1.toString().compareTo(sb2.toString());        }
Now, here is the strange problem I don't understand. I tested with my MR job. I know in the test data, I have 7 types data, 3 of them partitioned yearly, 4 of them partition daily. In the test data, for the 3 types partitioned daily, there are 2 days data of each type. So I expected the Input group count of the reducer should be 11, which is 4 x 2 + 3 = 11. In fact, if I don't use this custom MyPartitionKey, just use  Text as the key type,