我正在尝试执行一个简单的mapreduce程序，其中map接收输入，将其分成两部分（key=>string和value=>integer）。reducer每次都将得到classcastexception的对应键的值相加。我不明白，是什么在代码中造成了这个错误
我的代码：
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
public class Test {
public static class Map extends MapReduceBase implements
Mapper<LongWritable, Text, String, Integer> {
@Override
public void map(LongWritable key, Text value,
OutputCollector<String, Integer> output, Reporter reporter)
throws IOException {
String line = value.toString();
String[] lineParts = line.split(",");
output.collect(lineParts[0], Integer.parseInt(lineParts[1]));
}
}
public static class Reduce extends MapReduceBase implements
Reducer<String, Integer, String, Integer> {
@Override
public void reduce(String key, Iterator<Integer> values,
OutputCollector<String, Integer> output, Reporter reporter)
throws IOException {
int sum = 0;
while (values.hasNext()) {
sum = sum + values.next();
}
output.collect(key, sum);
}
}
public static void main(String[] args) throws Exception {
JobConf conf = new JobConf(Test.class);
conf.setJobName("ProductCount");
conf.setMapOutputKeyClass(String.class);
conf.setMapOutputValueClass(Integer.class);
conf.setOutputKeyClass(String.class);
conf.setOutputValueClass(Integer.class);
conf.setMapperClass(Map.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
}
}样本数据：
abc,10
abc,10
abc,10
def,9
def,9下面是堆栈跟踪。这和我的键值有关吗？
14/02/11 23:57:35 INFO mapred.JobClient: Task Id : attempt_201402110240_0013_m_000001_2, Status : FAILED
java.lang.ClassCastException: class java.lang.String
at java.lang.Class.asSubclass(Class.java:3018)
at org.apache.hadoop.mapred.JobConf.getOutputKeyComparator(JobConf.java:795)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.<init>(MapTask.java:816)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:382)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:324)
at org.apache.hadoop.mapred.Child$4.run(Child.java:268)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:396)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1115)
at org.apache.hadoop.mapred.Child.main(Child.java:262)
Exception in thread "main" java.io.IOException: Job failed!
at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1246)
at Test.main(Test.java:69)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.apache.hadoop.util.RunJar.main(RunJar.java:186)
1条答案
按热度按时间ui7jx7zq1#
在我看来，似乎您没有为输出使用正确的类。
从mapreduce教程之一：
键和值类必须可由框架序列化，因此需要实现可写接口。此外，关键类必须实现writablecomparable接口，以便通过框架进行排序。
因此，您应该更换
String.class与Text.class以及Integer.class与IntWritable.class.我希望这能解决你的问题。
为什么我不能使用基本的字符串或整数类？
integer和string实现了java的标准可序列化接口，如文档所示。问题是mapreduce序列化/反序列化值时没有使用这个标准接口，而是使用一个自己的接口（称为writable）。
那么他们为什么不直接使用基本的java接口呢？
简而言之：因为它更有效。可写接口在序列化时忽略类型定义，因为您已经在mapreduce代码中定义了输入/输出的类型。因为您的代码已经知道接下来会发生什么，所以不要像这样序列化字符串：
它可以序列化为：
如您所见，这节省了大量内存。
长话短说：读这篇很棒的博文。