NoSuchElementException while Reduce step

[Rajesh Sai T <ts...@gmail.com>]

Hi,

I'm new to Hadoop. I'm trying to write my custom data types for Writable
types. So, that Map class will produce my structure as value of key. And
Reduce class shall work on these list of my structure values. Below is my
program, please guide me what needs to be done to overcome this. I find it
does pass Map phase but while Reducing on last iteration is pops an
exception and job terminates.

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

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

public class InvertedGroupIndex {

public static class InvertedStruct implements Writable {
public String location;
public int count;
public InvertedStruct (int _count, String _str2) {
this.count = _count;
this.location = _str2;
}
public InvertedStruct () {
this(0, null);
}
/*public void set (String _str1, String _str2) {
this.word = _str1;
this.location = _str2;
}*/
public void write(DataOutput out) throws IOException {
out.writeInt(this.count);
out.writeChars(this.location);
}
public void readFields(DataInput in) throws IOException {
count = in.readInt();
location = in.readLine();
}
public String toString() {
return count + ";" + location;
}
public int getCount() {
return count;
}
public String getString() {
return location;
}
}
public static class InvertedMap extends MapReduceBase implements
Mapper<LongWritable, Text, Text, InvertedStruct> {
 private final static IntWritable count = new IntWritable(1);
private final static Text word = new Text();
 public void map (LongWritable key, Text val, OutputCollector<Text,
InvertedStruct>output, Reporter report) throws IOException {
FileSplit filesplit = (FileSplit) report.getInputSplit();
String fileName = filesplit.getPath().getName();
//location.set(fileName);
//InvertedStruct result = new InvertedStruct(1, fileName);
String line = val.toString();
StringTokenizer token = new StringTokenizer(line.toLowerCase());
while (token.hasMoreTokens()) {
word.set(token.nextToken());
output.collect(word, new InvertedStruct(1, fileName));
}
}
}

public static class InvertedReducer extends MapReduceBase implements
Reducer<Text, InvertedStruct, Text, Text> {
public void reduce(Text key, Iterator<InvertedStruct> values,
OutputCollector<Text, Text> output, Reporter reporter) throws IOException,
NoSuchElementException {
int sum=0;
StringBuilder toReturn = new StringBuilder();
while (values.hasNext()) {
sum += values.next().getCount();
toReturn.append(values.next().getString());
}
String s = String.valueOf(sum) + toReturn.toString();
output.collect(key, new Text(s));
}
}
public static void main (String[] args) throws IOException {
//JobClient client = new JobClient();
JobConf conf = new JobConf(InvertedGroupIndex.class);
conf.setJobName("InvertedGroupIndex");
conf.setMapperClass(InvertedMap.class);
//conf.setCombinerClass(InvertedReducer.class);
conf.setReducerClass(InvertedReducer.class);
conf.setMapOutputValueClass(InvertedStruct.class);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
 JobClient.runJob(conf);
}
}

Thanks,
Sai

Re: NoSuchElementException while Reduce step

[hadoop hive <ha...@gmail.com>]

hey there must be sum problem with the key or value, reducer didnt find the
expected value.

On Fri, Jan 27, 2012 at 1:23 AM, Rajesh Sai T <ts...@gmail.com> wrote:

> Hi,
>
> I'm new to Hadoop. I'm trying to write my custom data types for Writable
> types. So, that Map class will produce my structure as value of key. And
> Reduce class shall work on these list of my structure values. Below is my
> program, please guide me what needs to be done to overcome this. I find it
> does pass Map phase but while Reducing on last iteration is pops an
> exception and job terminates.
>
> import java.io.IOException;
> import java.util.*;
> import java.io.DataOutput;
> import java.io.DataInput;
>
> import org.apache.hadoop.fs.Path;
> import org.apache.hadoop.io.*;
> import org.apache.hadoop.mapred.*;
>
> public class InvertedGroupIndex {
>
> public static class InvertedStruct implements Writable {
> public String location;
> public int count;
> public InvertedStruct (int _count, String _str2) {
> this.count = _count;
> this.location = _str2;
> }
> public InvertedStruct () {
> this(0, null);
> }
> /*public void set (String _str1, String _str2) {
> this.word = _str1;
> this.location = _str2;
> }*/
> public void write(DataOutput out) throws IOException {
> out.writeInt(this.count);
> out.writeChars(this.location);
> }
> public void readFields(DataInput in) throws IOException {
> count = in.readInt();
> location = in.readLine();
> }
> public String toString() {
> return count + ";" + location;
> }
> public int getCount() {
> return count;
> }
> public String getString() {
> return location;
> }
> }
> public static class InvertedMap extends MapReduceBase implements
> Mapper<LongWritable, Text, Text, InvertedStruct> {
>  private final static IntWritable count = new IntWritable(1);
> private final static Text word = new Text();
>  public void map (LongWritable key, Text val, OutputCollector<Text,
> InvertedStruct>output, Reporter report) throws IOException {
> FileSplit filesplit = (FileSplit) report.getInputSplit();
> String fileName = filesplit.getPath().getName();
> //location.set(fileName);
> //InvertedStruct result = new InvertedStruct(1, fileName);
> String line = val.toString();
> StringTokenizer token = new StringTokenizer(line.toLowerCase());
> while (token.hasMoreTokens()) {
> word.set(token.nextToken());
> output.collect(word, new InvertedStruct(1, fileName));
> }
> }
> }
>
> public static class InvertedReducer extends MapReduceBase implements
> Reducer<Text, InvertedStruct, Text, Text> {
> public void reduce(Text key, Iterator<InvertedStruct> values,
> OutputCollector<Text, Text> output, Reporter reporter) throws IOException,
> NoSuchElementException {
> int sum=0;
> StringBuilder toReturn = new StringBuilder();
> while (values.hasNext()) {
> sum += values.next().getCount();
> toReturn.append(values.next().getString());
> }
> String s = String.valueOf(sum) + toReturn.toString();
> output.collect(key, new Text(s));
> }
> }
> public static void main (String[] args) throws IOException {
> //JobClient client = new JobClient();
> JobConf conf = new JobConf(InvertedGroupIndex.class);
> conf.setJobName("InvertedGroupIndex");
> conf.setMapperClass(InvertedMap.class);
> //conf.setCombinerClass(InvertedReducer.class);
> conf.setReducerClass(InvertedReducer.class);
> conf.setMapOutputValueClass(InvertedStruct.class);
> conf.setOutputKeyClass(Text.class);
> conf.setOutputValueClass(Text.class);
> FileInputFormat.setInputPaths(conf, new Path(args[0]));
> FileOutputFormat.setOutputPath(conf, new Path(args[1]));
>  JobClient.runJob(conf);
> }
> }
>
> Thanks,
> Sai
>