在 Reducer 中从 HBase 读取数据

Question

我是 Hadoop 和 HBase 的新手。让我用一个例子来解释我的问题。为简洁起见，数据很小。

假设我们有一个名为 item.log 的文件，它包含以下信息。

ITEM-1,PRODUCT-1
ITEM-2,PRODUCT-1
ITEM-3,PRODUCT-2
ITEM-4,PRODUCT-2
ITEM-5,PRODUCT-3
ITEM-6,PRODUCT-1
ITEM-7,PRODUCT-1
ITEM-8,PRODUCT-2
ITEM-9,PRODUCT-1

我有一个 map reduce 代码如下，

package org.sanjus.hadoop;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

public class ProductMapReduce {

    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> {
        
        public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException {
            String[] columns = value.toString().split(",");
            
            if (columns.length != 2) {
                System.out.println("Bad line/value " + value);
                return;
            }
                       
            Text word = new Text(columns[1]);
            LongWritable counter = new LongWritable(1L);
            
            output.collect(word, counter);
        }
    }
    
    
    public static class Reduce extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable> {

        public void reduce(Text key, Iterator<LongWritable> iterator, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException {
            long sum = 0L;
            
            while (iterator.hasNext()) {
                sum += iterator.next().get();
            }
            output.collect(key, new LongWritable(sum));
        }
        
    }
    
    public static void main(String[] args) throws IOException {
        JobConf conf = new JobConf(ProductMapReduce.class);
        conf.setJobName("Product Analyzer");
        
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(LongWritable.class);
        
        conf.setMapperClass(Map.class);
        conf.setCombinerClass(Reduce.class);
        conf.setReducerClass(Reduce.class);
        
        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);
        
        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));
        
        JobClient.runJob(conf);
    }
}

LABEL 1： map reduce 后的输出如下：

PRODUCT-1   5   
PRODUCT-2   3
PRODUCT-3   1

这是一个问题：

我在 HBase 中有一个表，其中包含以下信息。

PRODUCT-1   10$
PRODUCT-2   20$
PRODUCT-3   30$

问题/要求：我希望reduce阶段的输出作为“LABEL 1：”和上述HBase表中reduce输出的合并

PRODUCT-1   10$ * 5 = 50$
PRODUCT-2   20$ * 3 = 60$
PRODUCT-3   30$ * 1 = 30$

基本上，Key 是 PRODUCT-1，HBase 表中这个 key 的 value 是 10$，reducer 中相同 key 的 value 是 5，两个值相乘。 （$符号便于理解）

注意：我发现的示例基于 HBase 的输入或输出。我的场景是，输入和输出将是 HDFS 中的一个文件，而我需要使用 HBase 表中的信息处理 reducer 输出。

Answer 1

由于 HBase 支持高读取吞吐量，并且您只想读取 reducer 中的数据（将使用控制数量的数据）： 您可以使用 HBase API 根据 reducer 的键从表中读取数据。由于 Hbase 中的读取速度很快（约 10 毫秒，具体取决于获取的数据大小），我认为您的性能不会受到影响。 只需确保在 reducer 的 configure() 方法中初始化配置和 HTable。

Answer 2

这就是我所做的，

在我的 reducer 类中，我添加了重载方法 'setup'

private HTable htable;

private Configuration config;

protected void setup(Context context) throws IOException, InterruptedException {
    Configuration config = HBaseConfiguration.create();
    config.addResource(new Path("/etc/hbase/conf.hbase1/hbase-site.xml"));
    try {
        htable = new HTable(config, "MY_TABLE");
    }
    catch (IOException e) {
        System.out.println("Error getting table from HBase", e);
    }

}

使用 HTable.get api，我得到了 Result 对象。