我试图从另一个表(“mea\u data”)中收集的数据将值插入表(“mea\u interval”)。这个想法不是唯一的,它标识了一个数据类型。我使用measurewritable类对数据库进行读写,它实现了dbwritable和writable。当我运行jar时,我得到一个错误:
15/12/15 10:13:38 WARN mapred.LocalJobRunner: job_local957174264_0001
java.lang.ClassCastException: org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.io.NullWritable
at org.apache.phoenix.mapreduce.PhoenixRecordWriter.write(PhoenixRecordWriter.java:39)
at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:551)
at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:85)
at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.write(WrappedReducer.java:99)
at org.apache.hadoop.mapreduce.Reducer.reduce(Reducer.java:144)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:164)
at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:610)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:444)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:449))
我可以读取表mea\ U数据中的值。如果我在控制台中显示,它们看起来很好。我想这个错误发生在执行context.write的时候,但是我不明白为什么。
我附上了作业配置的代码和我的map类。如果你想看到我的代码的另一部分,不要犹豫。
事先谢谢。:)
作业配置:
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
public class Application {
public static void main(String[] args) {
final Configuration configuration = HBaseConfiguration.create();
final Job job;
try {
job = Job.getInstance(configuration, "phoenix-mr-job");
final String selectQuery = "SELECT * FROM \"mea_data\" where \"timestamp\" > 1450168200";
PhoenixMapReduceUtil.setInput(job, MeasureWritable.class, "mea_data", selectQuery);
// Set the target Phoenix table and the columns
PhoenixMapReduceUtil.setOutput(job, "\"mea_interval\"", "id_collection,startDate,endDate,value");
job.setMapperClass(MeasureMapper.class);
job.setReducerClass(MeasureReducer.class);
job.setOutputFormatClass(PhoenixOutputFormat.class);
// job.setInputFormatClass(PhoenixInputFormat.class);
job.setNumReduceTasks(10);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(MeasureWritable.class);
// TableMapReduceUtil.addDependencyJars(job);
job.waitForCompletion(true);
} catch (Exception e) {
e.printStackTrace();
}
}
}
Map器类:
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class MeasureMapper extends Mapper<NullWritable , MeasureWritable, LongWritable, Text> {
@Override
protected void map(NullWritable key, MeasureWritable measureWritable, Context context) throws IOException, InterruptedException {
final long timestamp = measureWritable.getTimestamp();
double val = measureWritable.getValue();
final long id = measureWritable.getId();
System.out.print("id : "+ new LongWritable(id));
System.out.print(" timestamp : "+ timestamp);
System.out.println(" val : "+ val);
try{
context.write(new LongWritable(id), new Text(timestamp + ";" + val));
} catch (Exception e) {
e.printStackTrace();
}
}
}
减速器等级:
import java.io.IOException;
import java.text.NumberFormat;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class MeasureReducer extends Reducer<LongWritable, Iterable<Text>, NullWritable, MeasureWritable> {
protected void reduce(LongWritable key, Iterable<Text> valeurs, Context context) throws IOException, InterruptedException {
MeasureWritable interval = new MeasureWritable();
interval.setId(Long.valueOf(key.toString()).longValue());
NumberFormat nf = NumberFormat.getInstance();
for(Text valeur : valeurs) {
String[] array = valeur.toString().split(";", -1);
interval.setStartingDate(Long.valueOf(array[0]).longValue());
interval.setEndingDate(Long.valueOf(array[0]).longValue());
try {
interval.setValue(nf.parse(array[1]).doubleValue() );
} catch (Exception e) {
e.printStackTrace();
}
}
context.write(NullWritable.get(), interval);
}
}
1条答案
按热度按时间yr9zkbsy1#
使用longwritable作为Map器的输入键以及Map方法的第一个参数,而不是nullwritable。