You are viewing a plain text version of this content. The canonical link for it is here.

Posted to common-user@hadoop.apache.org by xeonmailinglist-gmail <xe...@gmail.com> on 2015/04/01 16:54:22 UTC

can't set partition class to the configuration

Hi,

I have created a Mapper class[3] that filters out key values pairs that 
go to a specific partition. When I set the partition class in my code 
[1], I get the error in [2] and I don’t understand why this is 
happening. Any help to fix this?

[1]

|Configuration conf = cj.getConfiguration();
cj.setPartitionerClass(MyFilterMapper.class);
|

[2]

|The method setPartitionerClass(Class<? extends Partitioner>) in the type Job is not applicable for the arguments (Class<JobExecution.MyFilterMapper>)
|

[3]

|public static class MyFilterMapper
     extends Mapper<Object, Text, Text, IntWritable>{

         private Text word = new Text();
         private IntWritable rvalue = new IntWritable();

         public static final String REDUCE_TASK_REEXECUTE =
                 "mapreduce.reduce.task.reexecute";
         public static final int NULL_REDUCE_TASK = -1;

         private Class<? extends Partitioner<?, ?>> partitionerClass;
         private org.apache.hadoop.mapreduce.Partitioner<Object, Text> partitionerInstance;

         public void map(Object key, Text value, Context context
                 ) throws IOException, InterruptedException {
             Configuration conf = context.getConfiguration();
             partitionerInstance = new MyHashPartitioner<Object, Text>();

             int[] task_reexecute = conf.getInts(REDUCE_TASK_REEXECUTE);
             int nr_reduce_tasks = conf.getInt("mapreduce.job.reduces", 0);
             System.out.println("Tasks reexecute: " + task_reexecute + " NRREDUCETASKS: " + nr_reduce_tasks);
             StringTokenizer itr = new StringTokenizer(value.toString());
             while (itr.hasMoreTokens()) {
                 String wword = itr.nextToken();
                 Integer rrvalue = Integer.valueOf(itr.nextToken());
                 int partition = partitionerInstance.getPartition(wword, value, nr_reduce_tasks);

                 if (contains(partition, task_reexecute)) {
                     System.out.println("Partition Consumed: " + partition + " - key: " + key.toString() + " word: " + wword + " value - " + value.toString());
                     System.out.println("Partition Consumed: " + partition + " - word: " + wword + " value - ");// + rrvalue);

                     word.set(wword);
                     rvalue.set(rrvalue);
                     context.write(word, rvalue);
                 }
             }
         }

         public boolean contains(int partition, int[] set) {
             for(int i=0; i<set.length; i++){
                 if (partition == set[i])
                     return true;
             }

             return false;
         }
     }
|



-- 
--
Thanks,

Re: can't set partition class to the configuration

Posted by Ted Yu <yu...@gmail.com>.

The error message is very clear: a class which extends Partitioner is
expected.
Maybe you meant to specify MyHashPartitioner ?

Cheers

On Wed, Apr 1, 2015 at 7:54 AM, xeonmailinglist-gmail <
xeonmailinglist@gmail.com> wrote:

>  Hi,
>
> I have created a Mapper class[3] that filters out key values pairs that go
> to a specific partition. When I set the partition class in my code [1], I
> get the error in [2] and I don’t understand why this is happening. Any help
> to fix this?
>
> [1]
>
> Configuration conf = cj.getConfiguration();
> cj.setPartitionerClass(MyFilterMapper.class);
>
> [2]
>
> The method setPartitionerClass(Class<? extends Partitioner>) in the type Job is not applicable for the arguments (Class<JobExecution.MyFilterMapper>)
>
> [3]
>
> public static class MyFilterMapper
>     extends Mapper<Object, Text, Text, IntWritable>{
>
>         private Text word = new Text();
>         private IntWritable rvalue = new IntWritable();
>
>         public static final String REDUCE_TASK_REEXECUTE =
>                 "mapreduce.reduce.task.reexecute";
>         public static final int NULL_REDUCE_TASK = -1;
>
>         private Class<? extends Partitioner<?, ?>> partitionerClass;
>         private org.apache.hadoop.mapreduce.Partitioner<Object, Text> partitionerInstance;
>
>         public void map(Object key, Text value, Context context
>                 ) throws IOException, InterruptedException {
>             Configuration conf = context.getConfiguration();
>             partitionerInstance = new MyHashPartitioner<Object, Text>();
>
>             int[] task_reexecute = conf.getInts(REDUCE_TASK_REEXECUTE);
>             int nr_reduce_tasks = conf.getInt("mapreduce.job.reduces", 0);
>             System.out.println("Tasks reexecute: " + task_reexecute + " NRREDUCETASKS: " + nr_reduce_tasks);
>             StringTokenizer itr = new StringTokenizer(value.toString());
>             while (itr.hasMoreTokens()) {
>                 String wword = itr.nextToken();
>                 Integer rrvalue = Integer.valueOf(itr.nextToken());
>                 int partition = partitionerInstance.getPartition(wword, value, nr_reduce_tasks);
>
>                 if (contains(partition, task_reexecute)) {
>                     System.out.println("Partition Consumed: " + partition + " - key: " + key.toString() + " word: " + wword + " value - " + value.toString());
>                     System.out.println("Partition Consumed: " + partition + " - word: " + wword + " value - ");// + rrvalue);
>
>                     word.set(wword);
>                     rvalue.set(rrvalue);
>                     context.write(word, rvalue);
>                 }
>             }
>         }
>
>         public boolean contains(int partition, int[] set) {
>             for(int i=0; i<set.length; i++){
>                 if (partition == set[i])
>                     return true;
>             }
>
>             return false;
>         }
>     }
>
> 
>
> --
> --
> Thanks,
>
>