我需要使用hbase上的mr比较hbase中的两个表(表1、表2),并将摘要写入第三个表(表3)
我正在使用下面的tablemapreduceutil psuedo代码。Map器:表1:表3。
在mapper中,我需要比较table1和table2的值。在哪里示例化表2?
在Map器中,是否必须为每个Map器示例化table3?我想为整个mapreduce作业只示例化一次table3?
driver()
{
TableMapReduceUtil.initTableMapperJob(
table1, // input table
scan,
MyMapper.class, // mapper class
Text.class,
IntWritable.class,
job);
TableMapReduceUtil.initTableReducerJob(
table3, // output table
MyTableReducer.class,
job);
}
public static class MyMapper extends TableMapper<Text, IntWritable> {
public static final byte[] CF = "cf".getBytes();
public static final byte[] ATTR1 = "attr1".getBytes();
private final IntWritable ONE = new IntWritable(1);
private Text text = new Text();
public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
String val = new String(value.getValue(CF, ATTR1));
String diff;
//instantiate Table3 and compare with val. Do i have to instantiate for each mapper?
text.set(diff);
context.write(text, ONE);
}
}
public static class MyTableReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable> {
public static final byte[] CF = "cf".getBytes();
public static final byte[] COUNT = "count".getBytes();
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int i = 0;
for (IntWritable val : values) {
i += val.get();
}
Put put = new Put(Bytes.toBytes(key.toString()));
put.add(CF, COUNT, Bytes.toBytes(i));
context.write(null, put);
}
}
1条答案
按热度按时间euoag5mw1#
如果您试图在hbase中创建一个已经创建的表,它将抛出一个tableexistsexception,您可以选择忽略它。参见hbaseadmin文档。那么您就可以了-第一个创建表的Map器将实际创建它,然后其他Map器将抛出您将忽略的异常。