phoenix的MapReduce:org.apache.hadoop.io.LongWritable不能强制转换为or

Question

我正试图从另一个表（"mea_data"）中收集的数据中插入值到一个表中（"mea_interval"）。这个想法并不是唯一的，它确定了一种数据类型。我使用MeasureWritable类来读取和写入数据库，它实现了DBWritable和Writable。当我运行我的罐子时，我得到错误：

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_data中的值。如果我在控制台中显示，它们看起来不错。我认为这个错误发生在执行context.write的过程中，但我不明白为什么。

我附上了作业配置的代码和我的地图类。如果您想查看我代码的另一部分，请不要犹豫。

事先谢谢。：）

作业配置：

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();
        }
    }
}

映射程序类：

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);
    }
}

Answer 1

使用LongWritable作为Mapper的输入键以及映射方法的第一个参数，而不是NullWritable。