集成Hive和HBase

ReadFromTableMapper.java
//把hbase中的fruit表中的一部分数据，通过MR迁入到fruit_mr表, fruit_mr要先在hbase中创建好。
public class ReadFromTableMapper extends TableMapper<ImmutableBytesWritable, Put> {
    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        //1.获取rowkey//不可变的字节数组，按rowkey来读；scan扫描的封装到了result(1行里边的所有数据）
        byte[] rowkey = key.get();
        //2.创建put对象
        Put put = new Put(rowkey);  //创建put对象时要指定rowkey
        //3.添加列的信息
        for (Cell cell : value.rawCells()) { //获取这1列的所有cell
            if ("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
                if ("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
                    put.add(cell);
                }else if ("color".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
                    put.add(cell);
                }
            }
        }
        context.write(key, put);
    }
}


ReadFromTableReducer.java
// TableReducer<KEYIN, VALUEIN, KEYOUT> extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation>
//Mutation，Delete和Put都是它的子类；Hbase中每次操作可看做mutation这个类; Put已经封装了rowkey等所有数据
reducer初始化没有用到nullwritable
public class ReadFromTableReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
        for (Put put : values) {
            context.write(NullWritable.get(), put);
        }
    }
}


ReadFromTableDriver.jva
//实现Tool接口，执行任务时用ToolRunner调
public class ReadFromTableDriver implements Tool {
    private Configuration configuration = null;

    public int run(String[] strings) throws Exception { //获取任务并设置
        //1.获取job对象
        Job job = Job.getInstance(configuration);
        //2.设置job对象
        job.setJarByClass(ReadFromTableDriver.class);
        Scan scan = new Scan();//在这里传参数StartRow StopRow
        scan.setCacheBlocks(true); //设置缓存
        scan.setCaching(100); //扫100条数据再跟Hbase的server端交互，减少次数
        TableMapReduceUtil.initTableMapperJob("fruit", scan, //mapper去读数据;
                ReadFromTableMapper.class, ImmutableBytesWritable.class,//k v; fruit可通过args[0]传参
                Put.class, job);
        TableMapReduceUtil.initTableReducerJob("fruit_mr", //插入数据的表   //工具类初始化map，reduce
                ReadFromTableReducer.class, job);
        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1; //1表false
    }
    public void setConf(Configuration conf) { //hadoop的8个配置文件封装到Configuration中
        this.configuration = conf;
    }
    public Configuration getConf() {
        return configuration;
    }
    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new ReadFromTableDriver(), args); //最终调用的是上边的run方法；tool对象即ReadFromTableDriver的对象
        System.exit(run);//run为0或1
    }
}

打包运行任务
$ /opt/module/hadoop-2.7.2/bin/yarn jar hbase-0.0.1-SNAPSHOT.jar com.atguigu.mr.ReadFromTableDriver

public class ReadFromHdfsTableMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    private ImmutableBytesWritable k = new ImmutableBytesWritable();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1.获取1行
        String line = value.toString();
        //2.切割
        String[] fields = line.split("\t");
        //3.获取put对象
        String rowkey = fields[0];
        Put put = new Put(Bytes.toBytes(rowkey));
        //添加列信息| 列族/列名
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"),
                Bytes.toBytes(fields[1]));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("color"),
                Bytes.toBytes(fields[2]));
        context.write(k, put);
    }
}


public class ReadFromHdfsTableReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {
    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
        for (Put put : values) {
            context.write(NullWritable.get(), put);
        }
    }
}


public class ReadFromHdfsDriver implements Tool {
    private Configuration configuration;
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance();
        job.setJarByClass(ReadFromTableDriver.class);
        job.setMapperClass(ReadFromTableMapper.class);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Put.class);
        //设置文件路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        TableMapReduceUtil.initTableReducerJob(args[1], ReadFromTableReducer.class, job);
        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    }
    public void setConf(Configuration conf) {
        configuration =conf;
    }
    public Configuration getConf() {
        return configuration;
    }
    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new ReadFromTableDriver(), args); //传Tool对象
        System.exit(run);
    }
}

打包运行
$ /opt/module/hadoop-2.7.2/bin/yarn jar hbase-0.0.1-SNAPSHOT.jar com.atguigu.mr2.ReadFromHdfsDriver
提示：运行任务前，如果待数据导入的表不存在，则需要提前创建之。

/*

for read/write locks.</description>
</property>
<property>
  <name>hive.zookeeper.client.port</name>
  <value>2181</value>
  <description>The port of ZooKeeper servers to talk to. This is only needed for read/write locks.</description>
</property>

0: jdbc:hive2://hadoop101:10000> create table hive_hbase_emp_table(empno int, ename string, job string, mgr int, hiredate string, sal double, comm double, deptno int) stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 
with serdeproperties ("hbase.columns.mapping"=":key,info:ename,info:job,info:mgr,info:hiredate,info:sal,info:comm,info:deptno") 
tblproperties ("hbase.table.name"="hbase_emp_table");

完成之后，可以分别进入Hive和HBase查看，都生成了对应的表
    在Hive中创建临时中间表，用于load文件中的数据； 不能将数据直接load进Hive所关联HBase的那张表中；要经过mapreduce；
向hive_hbase_emp_table表中插入数据：
hive> insert into table hive_hbase_emp_table select * from emp;
查看Hive以及关联的HBase表中是否已经成功的同步插入了数据
Hive：
hive> select * from hive_hbase_emp_table;
HBase：
hbase> scan ‘hbase_emp_table’

创建外部表：
create external table relevance_hbase_emp(empno int, ename string, job string, mgr int, hiredate string, sal double, comm double, deptno int) stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 
with serdeproperties ("hbase.columns.mapping"=":key,info:ename,info:job,info:mgr,info:hiredate,info:sal,info:comm,info:deptno") 
tblproperties ("hbase.table.name"="hbase_emp_table");
关联后就可以使用Hive函数进行一些分析操作了
hive (default)> select * from relevance_hbase_emp;