我正在写一个mr作业,它将hbase表作为输入并转储到hdfs文件。我使用multipleinputs类(来自hadoop),因为我计划获取多个数据源。我编写了一个非常简单的mr程序(参见下面的源代码)。不幸的是,我遇到了以下错误:
java.lang.classcastexception:org.apache.hadoop.io.longwritable不能转换为org.apache.hadoop.hbase.io.immutablebyteswritable
我在伪分布式hadoop(1.2.0)和伪分布式hbase(0.95.1-hadoop1)上运行。
下面是完整的源代码:有趣的是:如果我注解掉multipleinputs行“multipleinputs.addinputpath(job,inputpath1,textinputformat.class,tablemap.class);”,先生的工作很顺利。
public class MixMR {
public static class TableMap extends TableMapper<Text, Text> {
public static final byte[] CF = "cf".getBytes();
public static final byte[] ATTR1 = "c1".getBytes();
public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
String key = Bytes.toString(row.get());
String val = new String(value.getValue(CF, ATTR1));
context.write(new Text(key), new Text(val));
}
}
public static class Reduce extends Reducer <Object, Text, Object, Text> {
public void reduce(Object key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
String ks = key.toString();
for (Text val : values){
context.write(new Text(ks), val);
}
}
}
public static void main(String[] args) throws Exception {
Path inputPath1 = new Path(args[0]);
Path outputPath = new Path(args[1]);
String tableName1 = "test";
Configuration config = HBaseConfiguration.create();
Job job = new Job(config, "ExampleRead");
job.setJarByClass(MixMR.class); // class that contains mapper
Scan scan = new Scan();
scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs
scan.setCacheBlocks(false); // don't set to true for MR jobs
scan.addFamily(Bytes.toBytes("cf"));
TableMapReduceUtil.initTableMapperJob(
tableName1, // input HBase table name
scan, // Scan instance to control CF and attribute selection
TableMap.class, // mapper
Text.class, // mapper output key
Text.class, // mapper output value
job);
job.setReducerClass(Reduce.class); // reducer class
job.setOutputFormatClass(TextOutputFormat.class);
// inputPath1 here has no effect for HBase table
MultipleInputs.addInputPath(job, inputPath1, TextInputFormat.class, TableMap.class);
FileOutputFormat.setOutputPath(job, outputPath);
job.waitForCompletion(true);
}
}
1条答案
按热度按时间q3qa4bjr1#
我得到了答案:在下面statement:replace textinputformat.class 到tableinputformat.class
multipleinputs.addinputpath(作业,inputpath1,textinputformat.class,tablemap.class);