专栏首页Jed的技术阶梯HBase Java API 03:HBase与MapReduce整合

HBase Java API 03:HBase与MapReduce整合

HBase版本:1.2.6

1. 案例说明

现有hbase表"student",其中内容如下:

hbase(main):025:0> scan 'student'
ROW     COLUMN+CELL                                                                            
 0001   column=info:age, timestamp=1516139523768, value=15                                     
 0001   column=info:name, timestamp=1516139523388, value=Madeline                              
 0002   column=info:age, timestamp=1516139523820, value=16                                     
 0002   column=info:name, timestamp=1516139523469, value=Jed                                   
 0003   column=info:age, timestamp=1516139523862, value=17                                     
 0003   column=info:name, timestamp=1516139523607, value=Olivia                                
 0004   column=info:age, timestamp=1516139523908, value=18                                     
 0004   column=info:name, timestamp=1516139523680, value=Jed                                   
 0005   column=info:age, timestamp=1516139525527, value=19                                     
 0005   column=info:name, timestamp=1516139523725, value=Sarah

需求:

编写MapReduce程序,把"student"表中"info"列族下的"name"那一列抽取出来,存入新HBase表"student_extract"中,要求"student_extract"表中只有"info"这个列族,"info"这个列族下只有"name"这个列

2. 代码实现

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

public class HBaseAndMapReduce {
    
    private static final String ZOOKEEPER_LIST = "repo:2181";
    private static final String INPUT_TABLE_NAME = "student";
    private static final String OUTPUT_TABLE_NAME = "student_extract";
    private static final byte[] FAMILY_NAME = "info".getBytes();
    private static final byte[] COLUMN_NAME = "name".getBytes();
    
    public static void main(String[] args) throws Exception {
        
        System.setProperty("HADOOP_USER_NAME", "root");
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", ZOOKEEPER_LIST);
        
        // 创建需要添加数据的新表
        Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();
        
        if(admin.tableExists(TableName.valueOf(OUTPUT_TABLE_NAME))) {
            admin.disableTable(TableName.valueOf(OUTPUT_TABLE_NAME));
            admin.deleteTable(TableName.valueOf(OUTPUT_TABLE_NAME));
        }
        
        HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(OUTPUT_TABLE_NAME));
        HColumnDescriptor columnDesc = new HColumnDescriptor(FAMILY_NAME);
        tableDesc.addFamily(columnDesc);
        admin.createTable(tableDesc);
        
        Job job = Job.getInstance(conf);
        
        // 初始化Job,设置Mapper和Reducer相关的参数
        Scan scan = new Scan();
        scan.addColumn(FAMILY_NAME, COLUMN_NAME);
        // 最后的参数代表是否添加依赖的jar包,本地模式设置为false
        TableMapReduceUtil.initTableMapperJob(INPUT_TABLE_NAME, scan, HBaseAndMapreduceMapper.class, Text.class, NullWritable.class, job, false);
        TableMapReduceUtil.initTableReducerJob(OUTPUT_TABLE_NAME, HBaseAndMapreduceReducer.class, job);
        
        /*
         * 上面两行代码中已经包含了下面的这5行代码
         * job.setMapperClass(HBaseAndMapreduceMapper.class);
         * job.setReducerClass(HBaseAndMapreduceReducer.class);
         * job.setMapOutputKeyClass(Text.class);
         * job.setMapOutputValueClass(NullWritable.class);
         * job.setOutputKeyClass(ImmutableBytesWritable.class);
         * job.setOutputValueClass(Mutation.class);
         */
        
        boolean successful = job.waitForCompletion(true);
        System.exit(successful ? 0 : -1);
        
    }
    
    // public abstract class TableMapper<KEYOUT, VALUEOUT> extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {...}
    private static class HBaseAndMapreduceMapper extends TableMapper<Text, NullWritable> {

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context)
                throws IOException, InterruptedException {
            
            String rowkey = Bytes.toString(key.copyBytes());
            List<Cell> cells = value.listCells();
            Text keyOut = new Text();
            StringBuilder sb = new StringBuilder();
            
            for(Cell cell : cells) {
                String family = new String(CellUtil.cloneFamily(cell));
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                String rowValue = new String(CellUtil.cloneValue(cell));
                sb.append(rowkey + "\t" + family + "\t" + qualifier + "\t" + rowValue);
                keyOut.set(sb.toString());
                context.write(keyOut, NullWritable.get());
                
                sb = new StringBuilder();
            }
        }
        
    }
    
    // public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT> extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation> {...}
    private static class HBaseAndMapreduceReducer extends TableReducer<Text, NullWritable, ImmutableBytesWritable> {

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            
            byte[] rowkey = Bytes.toBytes(key.toString().split("\t")[0]);
            byte[] family = Bytes.toBytes(key.toString().split("\t")[1]);
            byte[] qualifier = Bytes.toBytes(key.toString().split("\t")[2]);
            byte[] rowValue = Bytes.toBytes(key.toString().split("\t")[3]);
            
            ImmutableBytesWritable newRowKey = new ImmutableBytesWritable(rowkey);
            Put put = new Put(rowkey);
            put.addImmutable(family, qualifier, rowValue);
            
            context.write(newRowKey, put);
        }
        
    }

}

# 程序运行结束后,"student_extract"表中数据为:

hbase(main):033:0> scan 'student_extract'
ROW     COLUMN+CELL                                                                            
 0001   column=info:name, timestamp=1516142234629, value=Madeline                              
 0002   column=info:name, timestamp=1516142234629, value=Jed                                   
 0003   column=info:name, timestamp=1516142234629, value=Olivia                                
 0004   column=info:name, timestamp=1516142234629, value=Jed                                   
 0005   column=info:name, timestamp=1516142234629, value=Sarah

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Kafka 中使用 Avro 序列化框架(一):使用传统的 avro API 自定义序列化类和反序列化类

    关于 avro 的 maven 工程的搭建以及 avro 的入门知识,可以参考: Apache Avro 入门

    CoderJed
  • Spark读写HBase之使用Spark自带的API以及使用Bulk Load将大量数据导入HBase

    以上两个算子分别是基于Hadoop新版API和hadoop旧版API实现的,大部分代码都一样,需要注意的是新版API使用中Job类,旧版API使用JobConf...

    CoderJed
  • Kafka 中使用 Avro 序列化框架(二):使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化

    使用传统的 avro API 自定义序列化类和反序列化类比较麻烦,需要根据 schema 生成实体类,需要调用 avro 的 API 实现 对象到 byte[]...

    CoderJed
  • Hadoop Mapper 阶段将数据直接从 HDFS 导入 Hbase

    数据源格式如下: 20130512 1 -1 -1 13802 1 2013-05-12 07:26:22 20130512 1 -1 -1 13802 ...

    用户1177713
  • HBase新的客户端接口

    最近学习接触HBase的东西,看了《Habase in Action》,但里面关于HBase接口都是过时的接口,以下为HBase新的客户端接口:

    chaplinthink
  • Mapreduce和HBase新版本整合之WordCount计数案例

    先计数单词数量存到hdfs文件上,这个是以前的就做过的 package com.my.myhnase.mapreduce; import java.io.IO...

    汤高
  • HBase 2.0 协处理器实现 ES 数据同步

    在正式进行讲述实现之前,我觉得有必要说一下出发点。团队期初数据都是基于 HBase+Phoenix 这样架构进行持久化。随着业务的复杂性增加,对部分表的查询效率...

    迹_Jason
  • HBase整合MapReduce之建立HBase索引

    HBase索引主要用于提高Hbase中表数据的访问速度,有效的避免了全表扫描,HBase中的表根据行健被分成了多个Regions,通常一个region的一行都会...

    汤高
  • 大数据应用之HBase数据插入性能优化实测教程

    大家在使用HBase的过程中,总是面临性能优化的问题,本文从HBase客户端参数设置的角度,研究HBase客户端数据批量插入性能优化的问题。事实胜于雄辩,数据比...

    数据饕餮
  • 分布式存储系统-HBASE

    HBase –Hadoop Database,是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统,利用HBse技术可在廉价PC Server上搭建起大规模结...

    chaplinthink

扫码关注云+社区

领取腾讯云代金券