export HIVE_AUX_JARS_PATH=/opt/module/hbase-1.3.1/lib/*

<property>
  <name>hive.zookeeper.quorum</name>
  <value>hadoop101,hadoop102,hadoop103</value>
  <description>The list of ZooKeeper servers to talk to. This is only needed 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","hbase.mapred.output.outputtable" = "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’   或者 describe ‘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;

CREATE EXTERNAL TABLE test_hbase_1 (   
  id STRING,
  name STRING) 
STORED BY \'org.apache.hadoop.hive.hbase.HBaseStorageHandler\'
WITH SERDEPROPERTIES (\'hbase.table.name\'=\'test:table1\', \'hbase.mapred.output.outputtable\'=\'test:table1\',
\'hbase.columns.mapping\'=\':key,f:name\')

CREATE EXTERNAL TABLE test_hbase_2 (   
  id STRING,   
  age STRING) 
STORED BY \'org.apache.hadoop.hive.hbase.HBaseStorageHandler\'
WITH SERDEPROPERTIES (
\'hbase.table.name\'=\'test:table1\', 
\'hbase.mapred.output.outputtable\'=\'test:table1\',
\'hbase.columns.mapping\'=\':key,
f:age\');

drop table test_hbase_3;

CREATE EXTERNAL TABLE test_hbase_3 (   
  id STRING,
  name STRING,   
  age STRING) 
STORED BY \'org.apache.hadoop.hive.hbase.HBaseStorageHandler\'
WITH SERDEPROPERTIES (
\'hbase.table.name\'=\'test:table1\', 
\'hbase.mapred.output.outputtable\'=\'test:table1\',
\'hbase.columns.mapping\'=\':key,f:name,f:age\')

$ bin/hbase mapredcp

$ export HBASE_HOME=/opt/module/hbase
$ export HADOOP_HOME=/opt/module/hadoop-2.7.2
$ export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

在etc/hadoop下hadoop-env.sh中配置：（注意：在for循环之后配）,要重启下才生效
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/module/hbase/lib/*

-- 案例一：统计Student表中有多少行数据; 要在hbase中先有student表
$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar rowcounter student

-- 案例二：使用MapReduce将hdfs数据导入到HBase
1）在本地创建一个tsv格式的文件：fruit.tsv
   Apple    Red
   Pear        Yellow
   Pineapple    Yellow

2）创建HBase表
hbase(main):001:0> create \'fruit\',\'info\'

3）在HDFS中创建input_fruit文件夹并上传fruit.tsv文件
[kris@hadoop101 hadoop-2.7.2]$ hdfs dfs -mkdir /hbase/input_fruit  
[kris@hadoop101 datas]$ hadoop fs -put fruit.tsv /hbase/input_fruit

//把表名写死了;不写死可以通过命令行run(String[] args)传参
[kris@hadoop101 hadoop-2.7.2]$ bin/yarn jar HbaseTest-1.0-SNAPSHOT.jar com.atguigu.mr.ReadFromTableDriver
bin/yarn jar HbaseTest-1.0-SNAPSHOT.jar com.atguigu.mr2.ReadFromHdfsDriver /fruit.tsv fruit_mr //给它传参数

4）执行MapReduce到HBase的fruit表中
[kris@hadoop101 hadoop-2.7.2]$ bin/yarn jar /opt/module/hbase-1.3.1/lib/hbase-server-1.3.1.jar importtsv -Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit /hbase/input_fruit  //hdfs://hadoop101:9000/这个可省略


5）使用scan命令查看导入后的结果
hbase(main):001:0> scan ‘fruit’

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
提示：运行任务前，如果待数据导入的表不存在，则需要提前创建之。