我对hadoop和hbase还很陌生,我正在努力让它们一起工作。我已经建立了一个 .java
并获得 .jar
存档时没有任何错误。不管怎样,在启动程序时,我会遇到以下错误:
java.lang.Exception: java.io.IOException: wrong value class: class org.apache.hadoop.hbase.client.Put is not class org.apache.hadoop.io.IntWritable
at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:522)
Caused by: java.io.IOException: wrong value class: class org.apache.hadoop.hbase.client.Put is not class org.apache.hadoop.io.IntWritable
at org.apache.hadoop.mapred.IFile$Writer.append(IFile.java:194)
at org.apache.hadoop.mapred.Task$CombineOutputCollector.collect(Task.java:1378)
at org.apache.hadoop.mapred.Task$NewCombinerRunner$OutputConverter.write(Task.java:1695)
at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:89)
at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.write(WrappedReducer.java:105)
at TweetSentiment$ClassificationCounterReducer.reduce(TweetSentiment.java:131)
at TweetSentiment$ClassificationCounterReducer.reduce(TweetSentiment.java:114)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171)
at org.apache.hadoop.mapred.Task$NewCombinerRunner.combine(Task.java:1716)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1637)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.flush(MapTask.java:1489)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.close(MapTask.java:723)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:793)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:243)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:514)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1135)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:844)
我不明白为什么。
我已经尝试过一些不同版本的hadoop和hbase,但似乎不起作用。这是我的密码:
public static class ClassificationCounterReducer extends TableReducer<IntWritable,IntWritable,IntWritable> {
//private IntWritable result = new IntWritable();
public void reduce(IntWritable classification, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
Put put = new Put(Bytes.toBytes(classification.toString()));
put.addColumn( Bytes.toBytes("number"), Bytes.toBytes(""), Bytes.toBytes(sum) );
context.write(classification, put);
}
}
private static final String OUTPUT_TABLE = "sentiment";
public int run(String[] args) throws Exception {
Job job = Job.getInstance(getConf(), "Sentiment Count");
job.setJarByClass(TweetSentiment.class);
job.setMapperClass(ClassificatorMapper.class);
job.setCombinerClass(ClassificationCounterReducer.class);
job.setReducerClass(ClassificationCounterReducer.class);
TableMapReduceUtil.initTableReducerJob(
OUTPUT_TABLE,
TweetSentiment.ClassificationCounterReducer.class,
job);
job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new HBaseConfiguration(), new TweetSentiment(), args);
System.exit(res);
}
该代码预计将在twitter的tweets上运行情绪分析。Map程序应该从hdfs读取tweet,对它们进行分类,并将结果发送到合并器/还原器。reducer应该首先对它们进行计数,然后将它们存储在一个hbase表中,该表有两个条目,分别称为“情绪”和“数字”。
1条答案
按热度按时间n53p2ov01#
类型不匹配-正在输出
context.write(IntWritable, Put)
,但您的reducer类TableReducer<IntWritable,IntWritable,IntWritable>
.将第一行更改为:
此外,取消组合器,它不会工作。