import java.io.IOException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class CommonFriends {
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
private IntWritable friend = new IntWritable();
private Text friends = new Text();
public void map(Object key, Text value, Context context ) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString(),"\n");
while (itr.hasMoreTokens()) {
String[] line = itr.nextToken().split(" ");
if(line.length > 2 ){
int person = Integer.parseInt(line[0]);
for(int i=1; i<line.length;i++){
int ifriend = Integer.parseInt(line[i]);
friends.set((person < ifriend ? person+"-"+ifriend : ifriend+"-"+person));
for(int j=1; j< line.length; j++ ){
if( i != j ){
friend.set(Integer.parseInt(line[j]));
context.write(friends, friend);
}
}
}
}
}
}
}
public static class IntSumReducer extends Reducer<Text,IntWritable,Text,Text> {
private Text result = new Text();
public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
HashSet<IntWritable> duplicates = new HashSet();
ArrayList<Integer> tmp = new ArrayList();
for (IntWritable val : values) {
if(duplicates.contains(val))
tmp.add(val.get());
else
duplicates.add(val);
}
result.set(tmp.toString());
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Common Friends");
job.setJarByClass(CommonFriends.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
错误:java.io.ioexception:错误的值class:class org.apache.hadoop.io.text不是class org.apache.hadoop.io.intwriteable at org.apache.hadoop.mapred.ifile$writer.append(ifile)。java:194)在org.apache.hadoop.mapred.task$combineoutputcollector.collect(task。java:1350)在org.apache.hadoop.mapred.task$newcombinerrunner$outputconverter.write(任务。java:1667)在org.apache.hadoop.mapreduce.task.taskInputInputContextImpl.write(taskInputInputContextImpl。java:89)在org.apache.hadoop.mapreduce.lib.reduce.wrappedreducer$context.write(wrappedreducer。java:105)在commonfriends$intsumreducer.reduce(commonfriends。java:51) 在commonfriends$intsumreducer.reduce(commonfriends。java:38)在org.apache.hadoop.mapreduce.reducer.run(reducer。java:171)在org.apache.hadoop.mapred.task$newcombinerrunner.combine(task。java:1688)在org.apache.hadoop.mapred.maptask$mapoutputbuffer.sortandspill(maptask。java:1637)在org.apache.hadoop.mapred.maptask$mapoutputbuffer.flush(maptask。java:1489)在org.apache.hadoop.mapred.maptask$newoutputcollector.close(maptask。java:723)在org.apache.hadoop.mapred.maptask.runnewmapper(maptask。java:793)在org.apache.hadoop.mapred.maptask.run(maptask。java:341)在org.apache.hadoop.mapred.yarnchild$2.run(yarnchild。java:164)在javax.security.auth.subject.doas(主题)中的java.security.accesscontroller.doprivileged(本机方法)。java:422)在org.apache.hadoop.security.usergroupinformation.doas(usergroupinformation。java:1657)在org.apache.hadoop.mapred.yarnchild.main(yarnchild。java:158)
这是我的代码,错误信息如下。有什么想法吗??我认为mapper和reducer的输出类配置中的问题是输入文件是文件中的数字列表。如有需要,将提供更多细节。这个程序在朋友之间寻找共同的朋友
2条答案
按热度按时间tp5buhyn1#
删除
job.setCombinerClass(IntSumReducer.class);
在你的代码中可以解决这个问题ghg1uchk2#
刚刚查看了一下您的代码,似乎您正在使用reducer代码作为组合器代码。
有件事你需要检查一下。
您的组合器代码将以
<Text, IntWritable>
合路器的输出为<Text, Text>
格式。那么你的减速机的输入格式是
< Text, Text>
但您已将reducer的输入指定为< Text, IntWritable >
,所以它抛出了错误。可以做两件事:
1) 您可以考虑更改减速机的输出类型。
2) 您可以考虑编写一个单独的组合器代码。