分布式NoSQL列存储数据库Hbase_MR集成Hbase：读写Hbase规则（九）-阿里云开发者社区

分布式NoSQL列存储数据库Hbase（九）

知识点01：课程回顾

简述Hbase中hbase:meta表的功能及存储内容

功能：记录表的元数据信息
内容

rowkey：Hbase中每张表的每个Region的名称
列

Region名称
Region范围：startKey,stopKey
Region所在的RegionServer地址

简述Hbase中数据写入流程

step1：客户端连接ZK，获取meta表所在的地址，读取meta表数据
step2：根据表名，获取当前要操作的表的所有region的信息

region名称前缀：表名，startKey

step3：根据Rowkey，判断具体操作哪个Region
step4：获取对应Region的地址，请求对应的RegionServer
step5：RegionServer接受请求，将数据写入Region，先写入WAL
step6：根据列族来判断写入哪个Store中

写入Store的memstore中

简述Hbase中数据读取流程

step1：客户端连接ZK，获取meta表所在的地址，读取meta表数据
step2：根据表名，获取当前要操作的表的所有region的信息

region名称前缀：表名，startKey

step3：根据Rowkey，判断具体操作哪个Region
step4：获取对应Region的地址，请求对应的RegionServer
step5：：RegionServer接受请求，从Region中读取数据

先读memstore
判断查询数据是否做了缓存，如果做了缓存：就读BlockCache
最后读StoreFile

如果开启了缓存，查询结果会放入BlockCache

简述LSM模型的流程设计

step1：不论什么数据操作：增删改，都只对内存进行操作

删除和修改都是写入操作来代替的
内存写入成功，就返回

顺序读写内存
顺序读写磁盘
随机读写内存：memStore，BlockCache
随机读写磁盘：StoreFile

step2：数据写入内存，达到一定阈值，会将内存的数据写入磁盘
step3：定期将所有小文件和并为大文件，加快检索的效率

简述Hbase中的Flush、Compaction、Split的功能

Flush：将memstore中的数据刷写到HDFS，变成StoreFile文件

2.0之前

memstore：单个memstore达到128M，就会Flush
所有的memstore总存储达到95%，就会触发整个RS的Flush

2.0之后

设置一个水位线：max(128 / 列族个数,16)
高于水位线的memstore：就会flush
低于水位线的memstore：不会flush
所有都低于，都flush

Compaction：用于将storefile文件进行合并，并且删除过期数据【被标记为更新和删除的数据】

minor compact：轻量级合并，将最早的几个小的storefile文件进行合并，不会删除过期数据
major compact：重量级合并，将所有的storefile合并为一个storefile，会删除过期数据

默认每7天执行一次

2.0版本开始：in-memory-compact：在memstore中将数据提前进行合并

none：不开启
basic：只合并，不删除过期数据
eager：合并并且删除过期数据
adapter：合并，根据数据量来判断是否自动删除过期数据

Split：为了避免一个Region存储的数据量过大，导致负载过高，通过Split将一个region分为两个region，分摊负载

0.94之前：判断region中存储的文件大小是否达到10GB
2.0之前：根据Region个数，来计算划分的条件，达到4个以后，都是按照10GB来分

min(10GB,256 * region个数3次方)

2.0之后：根据region的个数做了判断

region的个数为1个：256M来划分
region的个数超过1个：10GB来划分

知识点02：课程目标

MapReduce读写Hbase

重点：记住读写的规则

Spark中读写Hbase规则与MapReduce的规则是一模一样的

应用：一般在工作中都是使用Spark来读写Hbase，如果是MapReduce可以使用Hive来实现

BulkLoad的实现【了解】

问题：大量的数据并发往Hbase中写入，会导致内存和磁盘的利用率非常高，会影响其他程序的性能
Hbase中提供两种写入数据的方式

Put：直接写入memstore
BulkLoad：先将数据转换为storefile文件，将storefile文件直接放入Hbase表的目录中

实现方式

自己开发代码
使用Hbase中的工具类来实现

协处理的介绍【了解】

什么是协处理器，分类
怎么开发协处理器：自己开发协处理器，实现索引表与原表数据同步

Hbase中的优化方案【重点：记住】

对于Hbase做了哪些性能的优化？
内存优化
压缩优化
参数优化
……

知识点03：MR集成Hbase：读Hbase规则

目标

掌握MapReduce中读取Hbase的开发规则

分析

读取由InputFormat决定

TextInputFormat：读取文件中的内容，每一行返回一个KV

K：行的偏移量：LongWritable
V：行的内容值：Text

TableInputFormat：负责实现读取Hbase的数据，将每个Rowkey的数据转换为一个KV对象

K：Rowkey的字节对象：ImmutableBytesWritable
V：Rowkey的数据内容：Result

实现

step1：调用工具类方法，初始化Input和Map

MapReduce中封装了工具类，实现读取Hbase数据

TableMapReduceUtil.initTableMapperJob

public static void initTableMapperJob(
      String table, 
      Scan scan,
      Class<? extends TableMapper> mapper,
      Class<?> outputKeyClass,
      Class<?> outputValueClass, 
      Job job
);

step2：构建Map类继承TableMapper类

/**
 * Extends the base <code>Mapper</code> class to add the required input key
 * and value classes.
 *
 * @param <KEYOUT>  The type of the key.
 * @param <VALUEOUT>  The type of the value.
 * @see org.apache.hadoop.mapreduce.Mapper
 */
@InterfaceAudience.Public
public abstract class TableMapper<KEYOUT, VALUEOUT>
extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {
}

总结

MapReduce读取Hbase数据的API已经封装好了，只需要调用工具类实现即可

知识点04：MR集成Hbase：读Hbase实现

目标

实现从Hbase读取数据，将数据写入文件中

分析

step1：使用TableInputFormat读取Hbase数据
step2：使用TextOutputFormat写入文件

实现

package bigdata.itcast.cn.hbase.mr;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
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.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;
/**
 * @ClassName ReadHbaseTable
 * @Description TODO 通过MapReduce读取Hbase表中的数据
 * @Create By     Frank
 */
public class ReadHbaseTable extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        //todo:1-创建
        Job job =  Job.getInstance(this.getConf(),"read");
        job.setJarByClass(ReadHbaseTable.class);
        //todo:2-配置
        //input&map
//        job.setInputFormatClass(TextInputFormat.class);
//        TextInputFormat.setInputPaths(job,new Path(""));
//        job.setMapperClass(null);
//        job.setMapOutputKeyClass(null);
//        job.setMapOutputValueClass(null);
        //input&map
        /**
         * public static void initTableMapperJob(
         *       String table,                              指定从哪张表读取
         *       Scan scan,                                 读取Hbase数据使用的Scan对象，自定义过滤器
         *       Class<? extends TableMapper> mapper,       Mapper类
         *       Class<?> outputKeyClass,                   Map输出的Key类型
         *       Class<?> outputValueClass,                 Map输出的Value类型
         *       Job job                                    当前的job
         *  )
         */
        //构建TableInputFormat用于读取Hbase的scan对象
        Scan scan = new Scan();//为了方便让你使用过滤器，提前过滤数据，再传递到MapReduce中，所以让你自定义一个scan对象
        //可以为scan设置过滤器,将过滤后的数据加载到MapReduce程序中
        TableMapReduceUtil.initTableMapperJob(
                "itcast:t1",
                scan,
                ReadHbaseMap.class,
                Text.class,
                Text.class,
                job
        );
        //reduce
        job.setNumReduceTasks(0);
        //output
        TextOutputFormat.setOutputPath(job,new Path("datas/output/hbase"));
        //todo:3-提交
        return job.waitForCompletion(true) ? 0:-1;
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        //指定Hbase服务端地址
        conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
        int status = ToolRunner.run(conf, new ReadHbaseTable(), args);
        System.exit(status);
    }
    /**
     * TableMapper<KEYOUT, VALUEOUT>
     * extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT>
     */
    public static class ReadHbaseMap extends TableMapper<Text, Text>{
        //rowkey
        Text outputKey = new Text();
        //每一列的数据
        Text outputValue = new Text();
        /**
         * 每个KV【一个Rowkey】调用一次map方法
         * @param key：rowkey
         * @param value：这个rowkey的数据
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            //给key进行赋值
            String rowkey = Bytes.toString(key.get());
            this.outputKey.set(rowkey);
            //给value赋值
            for(Cell cell : value.rawCells()){
                //得到每一列的数据
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String val  = Bytes.toString(CellUtil.cloneValue(cell));
                long ts = cell.getTimestamp();
                this.outputValue.set(family+"\t"+column+"\t"+val+"\t"+ts);
                //输出每一列的数据
                context.write(this.outputKey,this.outputValue);
            }
        }
    }
}

总结

最终也是调用了Hbase Java API
通过Scan来读取表的数据，返回到MapReduce程序汇总

知识点05：MR集成Hbase：写Hbase规则

目标

掌握MapReduce写入Hbase的开发规则

分析

输出由OutputFormat决定

TextOutputFormat：将KV输出写入文件中

TableOutputFormat：负责实现将上一步的KV数据写入Hbase表中

/**
 * Convert Map/Reduce output and write it to an HBase table. The KEY is ignored
 * while the output value <u>must</u> be either a {@link Put} or a
 * {@link Delete} instance.
 */
@InterfaceAudience.Public
public class TableOutputFormat<KEY> extends OutputFormat<KEY, Mutation>

要求输出的Value类型必须为Mutation类型：Put / Delete
Key是什么类型，不重要，在写入过程中，Key会被丢弃

实现

step1：调用工具类初始化Reduce和Output

MapReduce中封装了工具类，实现读取Hbase数据

TableMapReduceUtil.initTableReducerJob

/**
   * Use this before submitting a TableReduce job. It will
   * appropriately set up the JobConf.
   *
   * @param table  The output table.
   * @param reducer  The reducer class to use.
   * @param job  The current job to adjust.
   * @throws IOException When determining the region count fails.
   */
  public static void initTableReducerJob(
      String table,
      Class<? extends TableReducer> reducer,  指定Reduce类，不用传递Key和Value类型，因为Key不重要，Value定死了
      Job job
  );

step2：构建Reduce类继承TableReducer

/**
 * Extends the basic <code>Reducer</code> class to add the required key and
 * value input/output classes. 
 *
 * @param <KEYIN>  The type of the input key.
 * @param <VALUEIN>  The type of the input value.
 * @param <KEYOUT>  The type of the output key.
 * @see org.apache.hadoop.mapreduce.Reducer
 */
@InterfaceAudience.Public
public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>
  extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation> {
}

总结

MapReduce写入Hbase数据的API已经封装好了，只需要调用工具类实现即可

知识点06：MR集成Hbase：写Hbase实现

目标

实现从文件读取数据，将数据写入Hbase中

分析

step1：使用TextInputFormat读取文件中的数据
step2：构建Put对象，封装Rowkey以及列
step3：使用TableOutputFormat将数据写入Hbase表中

实现

Hbase中建表

create 'itcast:mrwrite','info'

实现

package bigdata.itcast.cn.hbase.mr;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;
/**
 * @ClassName WriteHbaseTable
 * @Description TODO 通过MapReduce将数据写入Hbase
 * @Create By     Frank
 */
public class WriteHbaseTable extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        //todo:1-创建
        Job job =  Job.getInstance(this.getConf(),"write");
        job.setJarByClass(WriteHbaseTable.class);
        //todo:2-配置
        //input
        TextInputFormat.setInputPaths(job,new Path("datas/hbase/writeHbase.txt"));
        //map
        job.setMapperClass(WriteToHbaseMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Put.class);
        //shuffle
        //reduce&output
        /**
         *  public static void initTableReducerJob(
         *     String table,                                将数据写入Hbase的哪张表
         *     Class<? extends TableReducer> reducer,       reducer的类
         *     Job job)                                     当前的job
         *
         *     以前输出的写法：
         *      job.setoutputKey：因为Key可以任意的，这里根本用不到
         *      job.setoutputValue：在TableReduce中将outputValue定死了，所以不用写
         *
         */
        TableMapReduceUtil.initTableReducerJob(
            "itcast:mrwrite",
            WriteToHbaseReduce.class,
            job
        );
        //output & reduce
//        job.setReducerClass(null);
//        job.setOutputKeyClass(null);
//        job.setOutputValueClass(null);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job,new Path(""));
        //todo:3-提交
        return job.waitForCompletion(true) ? 0:-1;
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
        int status = ToolRunner.run(conf, new WriteHbaseTable(), args);
        System.exit(status);
    }
    /**
     * 读取文件，将文件中的内容，id作为key，其他的每一列作为一个Put对象
     */
    public static class WriteToHbaseMap extends Mapper<LongWritable,Text,Text, Put>{
        Text rowkey = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //value：1 liudehua  18  male
            String[] split = value.toString().split("\t");
            String row = split[0];
            String name = split[1];
            String age = split[2];
            String sex = split[3];
            //将id作为rowkey，放在key中输出
            this.rowkey.set(row);
            //构造输出的Value
            Put putname = new Put(Bytes.toBytes(row));
            putname.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(name));
            context.write(rowkey,putname);
            Put putage = new Put(Bytes.toBytes(row));
            putage.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(age));
            context.write(rowkey,putage);
            Put putsex = new Put(Bytes.toBytes(row));
            putsex.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sex"),Bytes.toBytes(sex));
            context.write(rowkey,putsex);
        }
    }
    /**
     * public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>
     * extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation>
     *     最后Reduce输出的Value类型必须为Put类型，才能将数据写入Hbase
     */
    public static class WriteToHbaseReduce extends TableReducer<Text,Put,Text>{
        /**
         * 相同rowkey的所有Put都在一个迭代器中
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
            //直接遍历每个put对象，输出即可
            for (Put value : values) {
                context.write(key,value);
            }
        }
    }
}

总结

最终还是调用了Hbase Java API来实现的
通过构建Table对象，执行所有的Put对象实现将数据写入Hbase

知识点07：BulkLoad的介绍

目标

了解BulkLoad的功能及应用场景

分析

问题：有一批大数据量的数据，要写入Hbase中，如果按照传统的方案来写入Hbase，必须先写入内存，然后内存溢写到HDFS，导致Hbase的内存负载和HDFS的磁盘负载过高，影响业务
解决

写入Hbase方式
方式一：构建Put对象，先写内存
方式二：BulkLoad，直接将数据变成StoreFile文件，放入Hbase对应的HDFS目录中

数据不经过内存，读取数据时可以直接读取到

实现

step1：先将要写入的数据转换为HFILE文件
step2：将HFILE文件加载到Hbase的表中【直接将文件放入了Hbase表对应的HDFS目录中】

总结

应用场景：Hbase提供BulkLoad来实现大数据量不经过内存直接写入Hbase

特点

优点：不经过内存，降低了内存和磁盘的IO吞吐
缺点：性能上相对来说要慢一些，所有数据都不会在内存中被读取

知识点08：BulkLoad的实现

目标

实现BulkLoad方式加载数据到Hbase的表中

分析

step1：先将要写入的数据转换为HFILE文件
step2：将HFILE文件加载到Hbase的表中【直接将文件放入了Hbase表对应的HDFS目录中】

实现

开发代码
创建表

create 'mrhbase','info'

上传测试文件

hdfs dfs -mkdir -p  /bulkload/input
hdfs dfs -put writeHbase.txt /bulkload/input/

上传jar包到Linux上
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dQV6Ylwx-1616741489711)(20210326_分布式NoSQL列存储数据库Hbase（九）.assets/image-20210326103006352.png)]
step1：转换为HFILE

yarn jar bulkload.jar bigdata.itcast.cn.hbase.bulkload.TransHfileMR /bulkload/input/ /bulkload/output

运行找不到Hbase的jar包，手动申明HADOOP的环境变量即可，只在当前窗口有效

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/export/server/hbase-2.1.0/lib/shaded-clients/hbase-shaded-mapreduce-2.1.0.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/audience-annotations-0.5.0.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/commons-logging-1.2.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/findbugs-annotations-1.3.9-1.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/htrace-core4-4.2.0-incubating.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/log4j-1.2.17.jar:/export/server/hbase-2.1.0/lib/client-facing-thirdparty/slf4j-api-1.7.25.jar

step2：加载到Hbase表中

yarn jar bulkload.jar bigdata.itcast.cn.hbase.bulkload.BulkLoadToHbase /bulkload/output

总结

step1：先将数据转换为HFILE文件
step2：将HFILE加载到Hbase表中

知识点09：ImportTSV的使用

目标

了解ImportTSV工具的功能及使用
字面意思：导入tsv格式的数据文件

tsv：以制表符分隔每一列的文件
csv：以逗号分隔每一列的文件

分析

importtsv功能：将可以将任何一种结构化的文件导入Hbase的表中，【默认是使用Put方式来导入的】

默认导入tsv格式的文件

实现

使用方式一：直接使用Put方式导入

使用

yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar importtsv 
-Dimporttsv.columns=HBASE_ROW_KEY,cf1:name,cf1:age,cf2:sex <你要写入哪张表> <读取文件的文件地址>

手动指定分隔符

'-Dimporttsv.separator=,'，自己指定分隔符，默认分隔符为\t

举例

现在是数据

1 zhangsan 18 male
2 lisi 20 female

- 导入Hbase中
```shell
yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar  \
importtsv  \
-Dimporttsv.columns=a,b,c \ --指定表中的每一列与文件中的每一列的对应关系
<tablename>   \--指定导入哪张表
<inputdir>      \--指定导入哪个文件

yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar  \
importtsv  \
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:age,info:sex
mrhbase \
/bulkload/input

使用方式二：结合BulkLoad的方式来实现

step1：将普通文件转换为HFILE文件

yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar importtsv 
-Dimporttsv.columns=HBASE_ROW_KEY,cf1:name,cf1:age,cf2:sex 
-Dimporttsv.bulk.output=HFILE文件的存储地址
<你要写入哪张表> <读取文件的文件地址>

yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar
importtsv
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:age,info:sex
-Dimporttsv.bulk.output=/bulkload/output
mrhbase
/bulkload/input

- step2：使用bulkload加载数据

yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar completebulkload HFILE文件的地址表的名称

```shell
yarn jar /export/server/hbase-2.1.0/lib/hbase-mapreduce-2.1.0.jar completebulkload  /bulkload/output mrhbase

总结

importTSV：默认读取tsv格式

可以将数据直接写入Hbase
也可以将数据转换为HFILE文件

注意：importTSV可以读取别的格式：如果要读取csv格式

'-Dimporttsv.separator=,'

知识点10：协处理器的介绍

目标

了解协处理器的功能、分类和应用场景

分析

什么是协处理器？

协处理器指的是可以自定义开发一些功能集成到Hbase中
类似于Hive中的UDF，当没有这个功能时，可以使用协处理器来自定义开发，让Hbase支持对应的功能

实现

observer类：观察者类，类似于监听机制，MySQL中的触发器、Zookeeper中的监听

实现：监听A，如果A触发了，就执行B
监听对象

Region
Table
RegionServer
Master

触发：监听A，如果A触发了，执行B

pre：阻塞A，先执行B，再执行A
post：A先执行，B在A执行完成之后再执行

endpoint类：终端者类，类似于MySQL中的存储过程，Java中的方法

实现：固定一个代码逻辑，可以随时根据需求调用代码逻辑

总结

Hbase通过协处理器来弥补一些用户自定义功能的实现，例如二级索引等，但开发难度较高，一般通过第三方工具来实现

知识点11：协处理器的实现

目标

利用协处理器模拟实现二级索引同步原表与索引表数据

分析

step1：开发协处理器，监听原表的put请求
step2：拦截原表put请求，获取put操作，获取rowkey以及值
step3：构建索引表的rowkey，往索引表写入数据
step4：释放原表请求，往原表写入数据

实现

创建两张表

#rowkey：time_id
create 'proc1','info'
#rowkey：id_time
create 'proc2','info'

将开发好的协处理器jar包上传到hdfs上

hdfs dfs -mkdir -p /coprocessor/jar
mv bulkload.jar cop.jar
hdfs dfs -put cop.jar /coprocessor/jar/

添加协处理器到proc1中，用于监听proc1的操作

disable 'proc1'
alter 'proc1',METHOD => 'table_att','Coprocessor'=>'hdfs://node1:8020/coprocessor/jar/cop.jar|bigdata.itcast.cn.hbase.coprocessor.SyncCoprocessor|1001|'
enable 'proc1'

测试

put 'proc1','20191211_001','info:name','zhangsan'
scan 'proc1'
scan 'proc2'

卸载协处理器

disable 'proc1'
alter 'proc1',METHOD=>'table_att_unset',NAME=>'coprocessor$1'
enable 'proc1'

总结

协处理器API过于繁琐，基于不同的需求需要开发多个协处理器共同实现，不建议使用
建议使用Phoenix

知识点12：Hbase优化：内存分配

目标

了解Hbase中内存的管理及分配

分析

写缓存：Memstore
读缓存：BlockCache
使用的RegionServer的JVM堆内存

注意：所有使用JVM堆内存工具，都会有一个共同的问题：GC停顿
合理设计垃圾回收的机制来回收内存，避免GC停顿

实现

MemStore：写缓存

hbase.regionserver.global.memstore.size = 0.4

如果存多了，Flush到HDFS

BlockCache：读缓存

hfile.block.cache.size = 0.4

LRU淘汰算法，将最近最少被使用的数据从缓存中剔除

读多写少，降低MEMStore比例
读少写多，降低BlockCache比例

总结

可以根据实际的工作场景的需求，调整内存比例分配，提高性能

知识点13：Hbase优化：压缩机制

目标

了解Hbase中支持的压缩类型及配置实现

分析

Hbase的压缩源自于Hadoop对于压缩的支持
检查Hadoop支持的压缩类型

hadoop checknative

需要将Hadoop的本地库配置到Hbase中

实现

关闭Hbase的服务
配置Hbase的压缩本地库： lib/native/Linux-amd64-64

cd /export/server/hbase-2.1.0/
mkdir lib/native

将Hadoop的压缩本地库创建一个软链接到Hbase的lib/native目录下

ln -s /export/server/hadoop-2.7.5/lib/native /export/server/hbase-2.1.0/lib/native/Linux-amd64-64

启动Hbase服务

start-hbase.sh
hbase shell

创建表

create 'testcompress',{NAME=>'cf1',COMPRESSION => 'SNAPPY'}
put 'testcompress','001','cf1:name','laoda'

总结

Hbase提供了多种压缩机制实现对于大量数据的压缩存储，提高性能
压缩属于列族的属性：基于列族设计压缩

知识点14：Hbase优化：布隆过滤

目标

了解布隆过滤器的功能及使用

分析

什么是布隆过滤器？

是列族的一个属性，用于数据查询时对数据的过滤，类似于ORC文件中的布隆索引

实现

列族属性：BLOOMFILTER => NONE | ‘ROW’ | ROWCOL
NONE ：不开启布隆过滤器
ROW：开启行级布隆过滤

生成StoreFile文件时，会将这个文件中有哪些Rowkey的数据记录在文件的头部
当读取StoreFile文件时，会从文件头部或者这个StoreFile中的所有rowkey，自动判断是否包含需要的rowkey，如果包含就读取这个文件，如果不包含就不读这个文件

ROWCOL：行列级布隆过滤

生成StoreFile文件时，会将这个文件中有哪些Rowkey的以及对应的列族和列的信息数据记录在文件的头部
当读取StoreFile文件时，会从文件头部或者这个StoreFile中的所有rowkey以及列的信息，自动判断是否包含需要的rowkey以及列，如果包含就读取这个文件，如果不包含就不读这个文件

总结

Hbase通过布隆过滤器，在写入数据时，建立布隆索引，读取数据时，根据布隆索引加快数据的检索

知识点15：Hbase优化：列族属性

目标

了解其他常用列族属性

分析

{NAME => 'cf1', VERSIONS => '1', EVICT_BLOCKS_ON_CLOSE => 'false', NEW_VERSION_BEHAVIOR => 'false', KEEP_DELETED_CELLS => 'FALSE', CACHE_DATA_ON_WRITE => 'false', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', MIN_VERSIONS => '0', REPLICATION_SCOPE => '0', BLOOMFILTER => 'ROW', CACHE_INDEX_ON_WRITE => 'false', IN_MEMORY => 'false', CACHE_BLOOMS_ON_WRITE => 'false', PREFETCH_BLOCKS_ON_OPEN => 'false', COMPRESSION => 'SNAPPY', BLOCKCACHE => 'true', BLOCKSIZE => '65536'}

实现

NAME：表示列族的名称
VERSIONS：最大版本数

表示这个列族中的列最多可以存储几个版本的值

TTL：设置版本的自动过期时间，默认永不过期的，修改单位为秒

VERSIONS = 5
MIN_VERSIONS = 2
一旦到达TTL时间，会自动清理过期的版本，只保留2个版本

MIN_VERSIONS：最小版本数
BLOCKCACHE：开启缓存，如果列族开启了缓存，这个列族从HDFS的查询就会放入缓存中

默认就开启的
工作中要将不是经常读的列族的缓存关闭
使用LRU算法淘汰过期的数据

IN_MOMERY：最高缓存级别，meta表就是这个级别，一般情况下不建议开启

不会被优先淘汰

BLOCKSIZE：存储文件的块的大小
块越小，索引越多，查询越快，占用内存越多

块越大，索引越少，查询相对较慢，占用内存越少
一般不建议调整

知识点16：Hbase优化：其他优化

目标

了解Linux、HDFS、Zookeeper、Hbase其他属性优化

实现

Linux系统优化

开启文件系统的预读缓存可以提高读取速度

sudo blockdev --setra 32768 /dev/sda

最大限度使用物理内存

sudo sysctl -w vm.swappiness=0

调整文件及进程句柄数

sudo vi /etc/security/limits.conf 修改打开文件数限制
末尾添加：
*                soft    nofile          1024000
*                hard    nofile          1024000
Hive             -       nofile          1024000
hive             -       nproc           1024000 
$ sudo vi /etc/security/limits.d/20-nproc.conf 修改用户打开进程数限制
修改为：
#*          soft    nproc     4096
#root       soft    nproc     unlimited
*          soft    nproc     40960
root       soft    nproc     unlimited

HDFS优化

保证RPC调用会有较多的线程数

dfs.namenode.handler.count = 20
dfs.datanode.handler.count = 20

文件块大小的调整

dfs.blocksize = 256M

文件句柄数

dfs.datanode.max.transfer.threads = 4096

超时时间

dfs.image.transfer.timeout = 60000

避免DN错误宕机

dfs.datanode.failed.volumes.tolerated = 1

Zookeeper优化

优化Zookeeper会话超时时间

zookeeper.session.timeout = 90000

Hbase属性优化

设置RPC监听数量

hbase.regionserver.handler.count = 50

优化hbase客户端缓存

hbase.client.write.buffer = 2097152

指定scan.next扫描HBase所获取的行数

hbase.client.scanner.caching = 2147483647

附录一：Maven依赖

<repositories>
        <repository>
            <id>aliyun</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </repository>
    </repositories>
    <properties>
        <hadoop.version>2.7.3</hadoop.version>
        <hbase.version>2.1.2</hbase.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>${hbase.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-mapreduce</artifactId>
            <version>${hbase.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-auth</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies>

sion>
org.apache.hadoop
hadoop-mapreduce-client-jobclient
h a d o o p . v e r s i o n < / v e r s i o n > < / d e p e n d e n c y > < d e p e n d e n c y > < g r o u p I d > o r g . a p a c h e . h a d o o p < / g r o u p I d > < a r t i f a c t I d > h a d o o p − c o m m o n < / a r t i f a c t I d > < v e r s i o n > {hadoop.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>hadoop.version</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop−common</artifactId><version>{hadoop.version}
org.apache.hadoop
hadoop-mapreduce-client-core
h a d o o p . v e r s i o n < / v e r s i o n > < / d e p e n d e n c y > < d e p e n d e n c y > < g r o u p I d > o r g . a p a c h e . h a d o o p < / g r o u p I d > < a r t i f a c t I d > h a d o o p − a u t h < / a r t i f a c t I d > < v e r s i o n > {hadoop.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-auth</artifactId> <version>hadoop.version</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop−auth</artifactId><version>{hadoop.version}
org.apache.hadoop
hadoop-hdfs
${hadoop.version}

commons-io

2.6

分布式NoSQL列存储数据库Hbase_MR集成Hbase：读写Hbase规则（九）

分布式NoSQL列存储数据库Hbase（九）

知识点01：课程回顾

知识点02：课程目标

知识点03：MR集成Hbase：读Hbase规则

知识点04：MR集成Hbase：读Hbase实现

知识点05：MR集成Hbase：写Hbase规则

知识点06：MR集成Hbase：写Hbase实现

知识点07：BulkLoad的介绍

知识点08：BulkLoad的实现

知识点09：ImportTSV的使用

知识点10：协处理器的介绍

知识点11：协处理器的实现

知识点12：Hbase优化：内存分配

知识点13：Hbase优化：压缩机制

知识点14：Hbase优化：布隆过滤

知识点15：Hbase优化：列族属性

知识点16：Hbase优化：其他优化

附录一：Maven依赖

热门文章

最新文章

相关课程

相关电子书

相关实验场景