HBase详解

2024-03-18 08:16:46

文章目录

HBase简介

HBase – Hadoop Database，是一个高可靠性、高性能、面向列、可伸缩、实时读写的分布式数据库

利用Hadoop HDFS作为其文件存储系统,利用Hadoop MapReduce来处理HBase中的海量数据,利用Zookeeper作为其分布式协同服务，主要用来存储非结构化和半结构化的松散数据（列存 NoSQL 数据库）。

HBase特点

大：一个表可以有上亿行，上百万列。
面向列：面向列表（簇）的存储和权限控制，列（簇）独立检索。
稀疏：对于为空（NULL）的列，并不占用存储空间，因此，表可以设计的非常稀疏。
无模式：每一行都有一个可以排序的主键和任意多的列，列可以根据需要动态增加，同一张表中不同的行可以有截然不同的列。
数据多版本：每个单元中的数据可以有多个版本，默认情况下，版本号自动分配，版本号就是单元格插入时的时间戳。
数据类型单一：HBase中的数据都是字节数组，没有类型。

HBase架构

Master

为Region server分配region
负责Region server的负载均衡
发现失效的Region server并重新分配其上的region
管理用户对table的增删改操作

RegionServer

Region server维护region，处理对这些region的IO请求
Region server负责切分在运行过程中变得过大的region

Region

HBase自动把表水平划分成多个区域(region)，每个region会保存一个表里面某段连续的数据；每个表一开始只有一个region，随着数据不断插入表，region不断增大，当增大到一个阀值的时候，region就会等分会两个新的region（裂变）。
当table中的行不断增多，就会有越来越多的region。这样一张完整的表被保存在多个Regionserver 上。

Memstore&Storefile

一个region由多个store组成，一个store对应一个CF（列族）store包括位于内存中的memstore和位于磁盘的storefile写操作先写入memstore，当memstore中的数据达到某个阈值，hregionserver会启动flashcache进程写入storefile，每次写入形成单独的一个storefile
当storefile文件的数量增长到一定阈值后，系统会进行合并（minor、major compaction），在合并过程中会进行版本合并和删除工作（majar），形成更大的storefile
当一个region所有storefile的大小和数量超过一定阈值后，会把当前的region分割为两个，并由hmaster分配到相应的regionserver服务器，实现负载均衡
客户端检索数据，先在memstore找，找不到再找storefile

HLog

HLog文件就是一个普通的Hadoop Sequence File，Sequence File 的Key是HLogKey对象，HLogKey中记录了写入数据的归属信息，除了table和region名字外，同时还包括 sequence number和timestamp，timestamp是”写入时间”，sequence number的起始值为0，或者是最近一次存入文件系统中sequence number。
HLog SequeceFile的Value是HBase的KeyValue对象，即对应HFile中的KeyValue。

HBase数据模型

RowKey(行键)

唯一标识一行数据
可以通过RowKey获取一行数据
按照字典顺序排序的。
Row key只能存储64k的字节数据 10-100byte

ColumnFamily&Qualifier（列簇和列）

HBase表中的每个列都归属于某个列族，列族必须作为表模式(schema)定义的一部分预先给出。如 create ‘test’, ‘course’。
列名以列族作为前缀，每个“列族”都可以有多个列成员(column)；如course:math, course:english, 新的列族成员（列）可以随后按需、动态加入。
权限控制、存储以及调优都是在列族层面进行的；
HBase把同一列族里面的数据存储在同一目录下，由几个文件保存。

TimeStamp（时间戳）

在HBase每个cell存储单元对同一份数据有多个版本，根据唯一的时间戳来区分每个版本之间的差异，不同版本的数据按照时间倒序排序，最新的数据版本排在最前面。
时间戳的类型是 64位整型。
时间戳可以由HBase(在数据写入时自动)赋值，此时时间戳是精确到毫秒的当前系统时间
时间戳也可以由客户显式赋值，如果应用程序要避免数据版本冲突，就必须自己生成具有唯一性的时间戳。

Cell(存储单元)

由行和列的坐标交叉决定。
单元格是有版本的。
单元格的内容是未解析的字节数组。
由{row key， column( = +)， version} 唯一确定的单元。cell中的数据是没有类型的，全部是字节码形式存贮。

HBase读写流程

HBase Shell

HBaseAPI

常用java类

java类	HBase数据模型
Admin / HBaseAdmin/ HBaseConfiguration	数据库
HTable/HTableDescriptor	表
HColumnDescriptor	列簇
Put/Delete/Get/Scan/ResultScanner/	列
CellUtil	存储单元

示例代码

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Demo2API {
    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() {
        conf = HBaseConfiguration.create();
        // 其他配置参考http://hbase.apache.org/1.4/book.html#config.files
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建表
    @Test
    public void create_table() throws IOException {
        Admin admin = conn.getAdmin();
        // admin 相当于HBase的管理员
        // 创建表 传入表名(TableName.valueOf())
        HTableDescriptor tableName = new HTableDescriptor(TableName.valueOf("tableName"));
        // 创建列簇 传入列簇名
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        // 对列簇进行一些配置
        cf1.setMaxVersions(5); // 设置版本号
        cf1.setTimeToLive(30); // 设置TTL时间
        // 将创建好的列簇加入表
        tableName.addFamily(cf1);
        // 使用admin对象创建表
        admin.createTable(tableName);
    }

    // 删除表
    @Test
    public void drop_table() throws IOException {
        Admin admin = conn.getAdmin();
        String tableName = "tableName";
        admin.disableTable(TableName.valueOf(tableName));
        admin.deleteTable(TableName.valueOf(tableName));
    }

    // 添加一条数据
    @Test
    public void put() throws IOException {
        // 如果想要插入数据 查询数据 需要使用Table对象
        // 如果需要对表进行修改，获取表的一些配置、结构 使用HTableDescriptor对象
        Table testJavaAPI = conn.getTable(TableName.valueOf("testJavaAPI"));

        Put put = new Put("00001".getBytes());
        put.addColumn("cf1".getBytes(), "name".getBytes(), "zhangSan".getBytes());

        testJavaAPI.put(put);


    }

    // 获取一条数据
    @Test
    public void get() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("testJavaAPI"));
        Get get = new Get("00001".getBytes());
        Result rs = testJavaAPI.get(get);
        byte[] value = rs.getValue("cf1".getBytes(), "name".getBytes());
        System.out.println(Bytes.toString(value));
    }

    // 指定rowkey范围 扫描表
    @Test
    public void scan() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含startRow 不包含 endRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("007".getBytes());
        ResultScanner scanner = testJavaAPI.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {

            byte[] row = rs.getRow();// 获取rowkey
            String rk = Bytes.toString(row);
            System.out.println();
            if ("001".equals(rk)) {
                byte[] value = rs.getValue("cf1".getBytes(), "name".getBytes());
                System.out.println(Bytes.toString(value));
            } else if ("002".equals(rk)) {
                byte[] value = rs.getValue("cf1".getBytes(), "name0".getBytes());
                System.out.println(Bytes.toString(value));
                byte[] value1 = rs.getValue("cf1".getBytes(), "name1".getBytes());
                System.out.println(Bytes.toString(value1));
                byte[] value2 = rs.getValue("cf1".getBytes(), "name100".getBytes());
                System.out.println(Bytes.toString(value2));
                byte[] value3 = rs.getValue("cf1".getBytes(), "name2".getBytes());
                System.out.println(Bytes.toString(value3));
                byte[] value4 = rs.getValue("cf1".getBytes(), "name3".getBytes());
                System.out.println(Bytes.toString(value4));
                byte[] value5 = rs.getValue("cf1".getBytes(), "name4".getBytes());
                System.out.println(Bytes.toString(value5));
                byte[] value6 = rs.getValue("cf1".getBytes(), "name5".getBytes());
                System.out.println(Bytes.toString(value6));
            } else if ("007".equals(rk)) {
                byte[] value6 = rs.getValue("cf1".getBytes(), "name".getBytes());
                System.out.println(Bytes.toString(value6));
                byte[] value7 = rs.getValue("cf1".getBytes(), "age1".getBytes());
                System.out.println(Bytes.toString(value7));
            }

            rs = scanner.next();
        }
    }

    @Test
    public void cellUtil() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含startRow 不包含 endRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("007".getBytes());
        ResultScanner scanner = testJavaAPI.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            for (Cell cell : rs.listCells()) {
                byte[] rk = CellUtil.cloneRow(cell);
                byte[] cf = CellUtil.cloneFamily(cell);
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                byte[] value = CellUtil.cloneValue(cell);

                System.out.println("rowkey:" + Bytes.toString(rk) + ",columnsFamily:" + Bytes.toString(cf) + ",qualifier:" + Bytes.toString(qualifier) + ",value:" + Bytes.toString(value));
            }

            rs = scanner.next();
        }
    }

    // 读取文件并写入HBase
    @Test
    public void putAll() throws IOException {
        Admin admin = conn.getAdmin();
        // 判断表是否存在，不存在即创建
        if (!admin.tableExists(TableName.valueOf("students"))) {
            HTableDescriptor students = new HTableDescriptor(TableName.valueOf("students"));
            HColumnDescriptor info = new HColumnDescriptor("info");
            students.addFamily(info);
            admin.createTable(students);
        }

        Table students = conn.getTable(TableName.valueOf("students"));
        // 读取文件
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line = br.readLine();
        while (line != null) {
            String[] splits = line.split(",");
            String rk = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(Bytes.toBytes(rk));
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            students.put(put);

            line = br.readLine();
        }
        br.close();

    }

    @After
    public void closeAll() throws IOException {
        if (conn != null) {
            conn.close();
        }
    }
}

HBase过滤器

作用

过滤器的作用是在服务端判断数据是否满足条件，然后只将满足条件的数据返回给客户端
过滤器的类型很多，但是可以分为两大类：
- 比较过滤器：可应用于rowkey、列簇、列、列值过滤器
- 专用过滤器：只能适用于特定的过滤器

比较过滤器

比较运算符

LESS <
LESS_OR_EQUAL <=
EQUAL =
NOT_EQUAL <>
GREATER_OR_EQUAL >=
GREATER >
NO_OP 排除所有

常见的六大比较过滤器

BinaryComparator

按字节索引顺序比较指定字节数组，采用Bytes.compareTo(byte[])

BinaryPrefixComparator

通BinaryComparator，只是比较左端前缀的数据是否相同

NullComparator

判断给定的是否为空

BitComparator

按位比较

RegexStringComparator

提供一个正则的比较器，仅支持 EQUAL 和非EQUAL

SubstringComparator

判断提供的子串是否出现在中

示例代码

rowKey过滤器：RowFilter

通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来

    @Test
    // 通过RowFilter过滤比rowKey 1500100010 小的所有值出来
    public void BinaryComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes(1500100010));
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        ResultScanner scanner = students.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }

    }

列簇过滤器：FamilyFilter

通过FamilyFilter与SubstringComparator查询列簇名包含in的所有列簇下面的数据

    @Test
    // 通过FamilyFilter查询列簇名包含in的所有列簇下面的数据
    public void SubstringComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        SubstringComparator substringComparator = new SubstringComparator("in");
        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
        Scan scan = new Scan();
        scan.setFilter(familyFilter);
        ResultScanner scanner = students.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }

    }

通过FamilyFilter与 BinaryPrefixComparator 过滤出列簇以info开头的列簇下的所有数据

// 通过FamilyFilter与 BinaryPrefixComparator 过滤出列簇以info开头的所有列簇下的所有数据
    @Test
    public void BinaryPrefixComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));

        // 二进制前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("info".getBytes());
        // FamilyFilter 作用于列簇的过滤器
        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        Scan scan = new Scan();

        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100011".getBytes());
        // 通过setFilter方法设置过滤器
        scan.setFilter(familyFilter);

        ResultScanner scanner = students.getScanner(scan);

        printRS(scanner);

    }

列过滤器：QualifierFilter

通过QualifierFilter与SubstringComparator查询列名包含in的列的值

    public void printRS(ResultScanner scanner) throws IOException {
        for (Result rs : scanner) {
            String rowkey = Bytes.toString(rs.getRow());
            System.out.println("当前行的rowkey为：" + rowkey);
            for (Cell cell : rs.listCells()) {
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                byte[] bytes = CellUtil.cloneValue(cell);
                if ("age".equals(qualifier)) {
                    int value = Bytes.toInt(bytes);
                    System.out.println(family + ":" + qualifier + "的值为" + value);
                } else {
                    String value = Bytes.toString(bytes);
                    System.out.println(family + ":" + qualifier + "的值为" + value);
                }
            }
        }
    }

    @Test
    // 通过FamilyFilter查询列簇名包含in的所有列簇下面的数据
    public void SubstringComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        SubstringComparator substringComparator = new SubstringComparator("in");
        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
        Scan scan = new Scan();
        scan.setFilter(familyFilter);
        ResultScanner scanner = students.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }

    }

过滤出列的名字中包含 “am” 所有的列及列的值

    // 过滤出 列的名字 中 包含 "am" 所有的列 及列的值
    @Test
    public void SubstringComparatorQualifierFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));

        SubstringComparator substringComparator = new SubstringComparator("am");
        // 作用在列名上的过滤器
        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
        Scan scan = new Scan();

        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100011".getBytes());
        // 通过setFilter方法设置过滤器
        scan.setFilter(qualifierFilter);

        ResultScanner scanner = students.getScanner(scan);

        printRS(scanner);

    }

列值过滤器：ValueFilter

通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 “张” 开头的学生

    @Test
    // 通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
    public void BinaryPrefixComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("张".getBytes());
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);
        Scan scan = new Scan();
        scan.setFilter(valueFilter);
        ResultScanner scanner = students.getScanner(scan);

        printRS(scanner);

    }

过滤出文科的学生，只会返回clazz列，其他列的数据不符合条件，不会返回

    // 过滤出文科的学生
    // 只会返回clazz列，其他列的数据不符合条件，不会返回
    @Test
    public void RegexStringComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));

        // 使用正则表达式比较器
        RegexStringComparator regexStringComparator = new RegexStringComparator("^文科.*");
        // ValueFilter 会返回符合条件的cell，并不会返回整条数据
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, regexStringComparator);

        Scan scan = new Scan();

        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100011".getBytes());
        // 通过setFilter方法设置过滤器
        scan.setFilter(valueFilter);

        ResultScanner scanner = students.getScanner(scan);

        printRS(scanner);
    }

专用过滤器

单列值过滤器：SingleColumnValueFilter

SingleColumnValueFilter会返回满足条件的cell所在行的所有cell的值（即会返回一行数据）

通过SingleColumnValueFilter与查询文科班所有学生信息

    @Test
    // 通过SingleColumnValueFilter与查询文科班所有学生信息
    public void RegexStringComparatorFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                new RegexStringComparator("^文科.*")
        );

        Scan scan = new Scan();
        scan.setFilter(singleColumnValueFilter);
        ResultScanner scanner = students.getScanner(scan);

        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }

    }

列值排除过滤器：SingleColumnValueExcludeFilter

与SingleColumnValueFilter相反，会排除掉指定的列，其他的列全部返回

通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息，最终不返回clazz列

    @Test
    // 通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息，最终不返回clazz列
    public void RegexStringComparatorExcludeFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                new BinaryComparator("文科一班".getBytes())
        );

        Scan scan = new Scan();
        scan.setFilter(singleColumnValueExcludeFilter);
        ResultScanner scanner = students.getScanner(scan);

        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            // clazz列为空
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }

    }

rowkey前缀过滤器：PrefixFilter

通过PrefixFilter查询以150010008开头的所有前缀的rowkey

    @Test
    // 通过PrefixFilter查询以150010008开头的所有前缀的rowkey
    public void PrefixFilterFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
        Scan scan = new Scan();
        scan.setFilter(prefixFilter);
        ResultScanner scanner = students.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            int age = Bytes.toInt(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            // clazz列为空
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz + "\t");

            rs = scanner.next();
        }
    }

分页过滤器PageFilter

通过PageFilter查询第三页的数据，每页10条
使用PageFilter分页效率比较低，每次都需要扫描前面的数据，直到扫描到所需要查的数据
可设计一个合理的rowkey来实现分页需求

    @Test
    // 通过PageFilter查询第三页的数据，每页10条
    public void PageFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        int PageNum = 3;
        int PageSize = 10;
        Scan scan = new Scan();
        if (PageNum == 1) {
            scan.withStartRow("".getBytes());
            //使用分页过滤器，实现数据的分页
            PageFilter pageFilter = new PageFilter(PageSize);
            scan.setFilter(pageFilter);
            ResultScanner scanner = students.getScanner(scan);
            printRS(scanner);
        } else {
            String current_page_start_rows = "";
            int scanDatas = (PageNum - 1) * PageSize + 1;
            PageFilter pageFilter = new PageFilter(scanDatas);
            scan.setFilter(pageFilter);
            ResultScanner scanner = students.getScanner(scan);
            for (Result rs : scanner) {
                current_page_start_rows = Bytes.toString(rs.getRow());
            }
            scan.withStartRow(current_page_start_rows.getBytes());
            PageFilter pageFilter1 = new PageFilter(PageSize);
            scan.setFilter(pageFilter1);
            ResultScanner scanner1 = students.getScanner(scan);
            printRS(scanner1);

        }

    }

通过合理的设置rowkey来实现分页功能

    @Test
    // 通过合理的设置rowkey来实现分页功能，提高效率
    public void PageFilterTest2() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        int PageSize = 10;
        int PageNum = 3;

        int baseId = 1500100000;
        int start_row = baseId + (PageNum - 1) * PageSize + 1;
        int end_row = start_row + PageSize;
        Scan scan = new Scan();
        scan.withStartRow(String.valueOf(start_row).getBytes());
        scan.withStopRow(String.valueOf(end_row).getBytes());

        ResultScanner scanner = students.getScanner(scan);

        printRS(scanner);


    }

多过滤器综合查询

查询文科班中的学生中学号以150010008开头并且年龄小于23的学生信息

    @Test
    // 查询文科班中的学生中学号以150010008开头并且年龄小于23的学生信息
    public void FilterListFilter() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareFilter.CompareOp.EQUAL
                , new RegexStringComparator("^文科.*"));
        PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
        SingleColumnValueFilter singleColumnValueFilter1 = new SingleColumnValueFilter(
                "info".getBytes()
                , "age".getBytes()
                , CompareFilter.CompareOp.LESS
                , new BinaryComparator(Bytes.toBytes(23)));

        FilterList filterList = new FilterList();
        filterList.addFilter(singleColumnValueFilter);
        filterList.addFilter(prefixFilter);
        filterList.addFilter(singleColumnValueFilter1);
        scan.setFilter(filterList);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

Phoenix

Phoenix搭建

Phoenix 4.15 HBase 1.4.6 hadoop 2.7.6

关闭hbase集群，在master中执行

stop-hbase.sh

上传解压配置环境变量

解压

tar -xvf apache-phoenix-4.15.0-HBase-1.4-bin.tar.gz 

改名

mv apache-phoenix-4.15.0-HBase-1.4-bin phoenix-4.15.0

将phoenix-4.15.0-HBase-1.4-server.jar复制到所有节点的hbase lib目录下

scp /usr/local/soft/phoenix-4.15.0/phoenix-4.15.0-HBase-1.4-server.jar master:/usr/local/soft/hbase-1.4.6/lib/

scp /usr/local/soft/phoenix-4.15.0/phoenix-4.15.0-HBase-1.4-server.jar node1:/usr/local/soft/hbase-1.4.6/lib/

scp /usr/local/soft/phoenix-4.15.0/phoenix-4.15.0-HBase-1.4-server.jar node2:/usr/local/soft/hbase-1.4.6/lib/

启动hbase ，在master中执行

start-hbase.sh

配置环境变量

vim /etc/profile
增加：export PHOENIX_HOME=/usr/local/soft/phoenix-4.15.0
	path:$PHOENIX_HOME/bin
source /etc/profile

Phoenix使用

连接sqlline

sqlline.py master,node1,node2

# 出现
163/163 (100%) Done
Done
sqlline version 1.5.0
0: jdbc:phoenix:master,node1,node2>

常用命令

# 1、创建表

CREATE TABLE IF NOT EXISTS STUDENT (
 id VARCHAR NOT NULL PRIMARY KEY, 
 name VARCHAR,
 age BIGINT, 
 gender VARCHAR ,
 clazz VARCHAR
);

# 2、显示所有表
 !table

# 3、插入数据
upsert into STUDENT values('1500100004','葛德曜',24,'男','理科三班');
upsert into STUDENT values('1500100005','宣谷芹',24,'男','理科六班');
upsert into STUDENT values('1500100006','羿彦昌',24,'女','理科三班');


# 4、查询数据,支持大部分sql语法，
select * from STUDENT ;
select * from STUDENT where age=24;
select gender ,count(*) from STUDENT group by gender;
select * from student order by gender;

# 5、删除数据
delete from STUDENT where id='1500100004';


# 6、删除表
drop table STUDENT;
 
 
# 7、退出命令行
!quit

更多语法参照官网
https://phoenix.apache.org/language/index.html#upsert_select

phoenix映射

默认情况下，直接在hbase中创建的表，通过phoenix是查看不到的
如果需要在phoenix中操作直接在hbase中创建的表，则需要在phoenix中进行表的映射。映射方式有两种：视图映射和表映射

视图映射

Phoenix创建的视图是只读的，所以只能用来做查询，无法通过视图对源数据进行修改等操作

# hbase shell 进入hbase命令行
hbase shell 

# 创建hbase表
create 'test','name','company' 

# 插入数据
put 'test','001','name:firstname','zhangsan'
put 'test','001','name:lastname','zhangsan'
put 'test','001','company:name','数加'
put 'test','001','company:address','合肥'


# 在phoenix创建视图， primary key 对应到hbase中的rowkey

create view "test" (
empid varchar primary key,
"name"."firstname" varchar,
"name"."lastname"  varchar,
"company"."name"  varchar,
"company"."address" varchar
);



CREATE view "students" (
 id VARCHAR NOT NULL PRIMARY KEY, 
 "info"."name" VARCHAR,
 "info"."age" VARCHAR, 
 "info"."gender" VARCHAR ,
 "info"."clazz" VARCHAR
) column_encoded_bytes=0;

# 在phoenix查询数据，表名通过双引号引起来
select * from "test";

# 删除视图
drop view "test";

表映射

使用Apache Phoenix创建对HBase的表映射，有两类：

1）当HBase中已经存在表时，可以以类似创建视图的方式创建关联表，只需要将create view改为create table即可。

2）当HBase中不存在表时，可以直接使用create table指令创建需要的表，并且在创建指令中可以根据需要对HBase表结构进行显示的说明。

第1）种情况下，如在之前的基础上已经存在了test表，则表映射的语句如下：

create table "test" (
empid varchar primary key,
"name"."firstname" varchar,
"name"."lastname"varchar,
"company"."name"  varchar,
"company"."address" varchar
)column_encoded_bytes=0;

upsert into  "test"  values('1','2','3','4','5');

CREATE table  "students" (
 id VARCHAR NOT NULL PRIMARY KEY, 
 "info"."name" VARCHAR,
 "info"."age" VARCHAR, 
 "info"."gender" VARCHAR ,
 "info"."clazz" VARCHAR
) column_encoded_bytes=0;

upsert into "students" values('1500110004','葛德曜','24','男','理科三班');

使用create table创建的关联表，如果对表进行了修改，源数据也会改变，同时如果关联表被删除，源表也会被删除。但是视图就不会，如果删除视图，源数据不会发生改变。

Phoenix二级索引

建立行键与列值的映射关系
对于Hbase，如果想精确定位到某行记录，唯一的办法就是通过rowkey查询。如果不通过rowkey查找数据，就必须逐行比较每一行的值，对于较大的表，全表扫描的代价是不可接受的。

开启索引支持

# 关闭hbase集群
stop-hbase.sh

# 在/usr/local/soft/hbase-1.4.6/conf/hbase-site.xml中增加如下配置

<property>
  <name>hbase.regionserver.wal.codec</name>
  <value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value>
</property>
<property>
    <name>hbase.rpc.timeout</name>
    <value>60000000</value>
</property>
<property>
    <name>hbase.client.scanner.timeout.period</name>
    <value>60000000</value>
</property>
<property>
    <name>phoenix.query.timeoutMs</name>
    <value>60000000</value>
</property>


# 同步到所有节点
scp hbase-site.xml node1:`pwd`
scp hbase-site.xml node2:`pwd`

# 修改phoenix目录下的bin目录中的hbase-site.xml
<property>
    <name>hbase.rpc.timeout</name>
    <value>60000000</value>
</property>
<property>
    <name>hbase.client.scanner.timeout.period</name>
    <value>60000000</value>
</property>
<property>
    <name>phoenix.query.timeoutMs</name>
    <value>60000000</value>
</property>


# 启动hbase
start-hbase.sh
# 重新进入phoenix客户端
sqlline.sql master,node1,node2

创建索引

全局索引

全局索引适合读多写少的场景。如果使用全局索引，读数据基本不损耗性能，所有的性能损耗都来源于写数据。数据表的添加、删除和修改都会更新相关的索引表（数据删除了，索引表中的数据也会删除；数据增加了，索引表的数据也会增加）

注意: 对于全局索引在默认情况下，在查询语句中检索的列如果不在索引表中，Phoenix不会使用索引表将，除非使用hint。

# 创建DIANXIN.sql
CREATE TABLE IF NOT EXISTS DIANXIN (
     mdn VARCHAR ,
     start_date VARCHAR ,
     end_date VARCHAR ,
     county VARCHAR,
     x DOUBLE ,
     y  DOUBLE,
     bsid VARCHAR,
     grid_id  VARCHAR,
     biz_type VARCHAR, 
     event_type VARCHAR , 
     data_source VARCHAR ,
     CONSTRAINT PK PRIMARY KEY (mdn,start_date)
) column_encoded_bytes=0;

# 上传数据DIANXIN.csv

# 导入数据
psql.py master,node1,node2 DIANXIN.sql DIANXIN.csv

# 创建全局索引
CREATE INDEX DIANXIN_INDEX ON DIANXIN ( end_date );

# 查询数据 ( 索引未生效)
select * from DIANXIN where end_date = '20180503154014';

# 强制使用索引 （索引生效） hint
select /*+ INDEX(DIANXIN DIANXIN_INDEX) */  * from DIANXIN where end_date = '20180503154014';

select /*+ INDEX(DIANXIN DIANXIN_INDEX) */  * from DIANXIN where end_date = '20180503154014'  and start_date = '20180503154614';

# 取索引列，（索引生效）
select end_date from DIANXIN where end_date = '20180503154014';

# 创建多列索引
CREATE INDEX DIANXIN_INDEX1 ON DIANXIN ( end_date,COUNTY );

# 多条件查询 （索引生效）
select end_date,MDN,COUNTY from DIANXIN where end_date = '20180503154014' and COUNTY = '8340104';

# 查询所有列 (索引未生效)
select  * from DIANXIN where end_date = '20180503154014'  and COUNTY = '8340104';

# 查询所有列 （索引生效）
select /*+ INDEX(DIANXIN DIANXIN_INDEX1) */ * from DIANXIN where end_date = '20180503154014' and COUNTY = '8340104';

# 单条件  (索引未生效)
select end_date from DIANXIN where  COUNTY = '8340103';

本地索引

本地索引适合写多读少的场景，或者存储空间有限的场景。和全局索引一样，Phoenix也会在查询的时候自动选择是否使用本地索引。本地索引因为索引数据和原数据存储在同一台机器上，避免网络数据传输的开销，所以更适合写多的场景。由于无法提前确定数据在哪个Region上，所以在读数据的时候，需要检查每个Region上的数据从而带来一些性能损耗。

注意:对于本地索引，查询中无论是否指定hint或者是查询的列是否都在索引表中，都会使用索引表。

@ 创建本地索引
CREATE LOCAL INDEX DIANXIN_LOCAL_IDEX ON DIANXIN(grid_id);

# 索引生效
select grid_id from dianxin where grid_id='117285031820040';

# 索引生效
select * from dianxin where grid_id='117285031820040';

全局索引与本地索引区别

全局索引：读多写少，会单独建立索引表
本地索引：读少写多，索引数据和原数据保存在同一台机器上

覆盖索引

覆盖索引是把原数据存储在索引数据表中，这样在查询时不需要再去HBase的原表获取数据就，直接返回查询结果。

注意：查询是 select 的列和 where 的列都需要在索引中出现。

# 创建覆盖索引
CREATE INDEX DIANXIN_INDEX_COVER ON DIANXIN ( x,y ) INCLUDE ( county );

# 查询所有列 (索引未生效)
select * from dianxin where x=117.288 and y =31.822;

# 强制使用索引 (索引生效)
select /*+ INDEX(DIANXIN DIANXIN_INDEX_COVER) */ * from dianxin where x=117.288 and y =31.822;

# 查询索引中的列 (索引生效)
select mdn,x,y,county from dianxin where x=117.288 and y =31.822;

查询条件必须放在索引中  select 中的列可以放在INCLUDE （将数据保存在索引中）


select /*+ INDEX(DIANXIN DIANXIN_INDEX_COVER) */ x,y,count(*) from dianxin group by x,y;

PhoenixJDBC

# 导入依赖

<dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-core</artifactId>
    <version>4.15.0-HBase-1.4</version>
</dependency>

Connection conn = DriverManager.getConnection("jdbc:phoenix:master,node1,node2:2181");
Statement stat = conn.createStatement();
ResultSet rs = stat.executeQuery("select * from dianxin limit 10");
while(rs.next()){
	String mdn = rs.getString("mdn");
	System.out.println(mdn);
}
stat.close();
conn.close();

HBase SQL与Hive SQL执行流程

HBase的MapReduce过程

#### 代码示例

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
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.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

public class Demo7MapReduceReadAndWriteHBase {
    // 读取HBase中的students表，统计每个班级的人数，将结果写回HBase clazz_num表
    public static class ReadHBaseMapper extends TableMapper<Text, IntWritable> {
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            String rowkey = Bytes.toString(key.get());
            String clazz = Bytes.toString(value.getValue("info".getBytes(), "clazz".getBytes()));

            context.write(new Text(clazz), new IntWritable(1));
        }

    }

    public static class WriteHBaseReducer extends TableReducer<Text, IntWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable i : values) {
                sum += i.get();
            }
            // create 'clazz_num','cf1'
            Put put = new Put(key.toString().getBytes());
            put.addColumn("cf1".getBytes(), "num".getBytes(), Bytes.toBytes(sum));

            context.write(NullWritable.get(), put);

        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,node1,node2:2181");

        Job job = Job.getInstance(conf);
        job.setJobName(Demo7MapReduceReadAndWriteHBase.class.getName());
        job.setJarByClass(Demo7MapReduceReadAndWriteHBase.class);

        // 配置map任务
        // 使用HBase提供的TableMapReduceUtil工具进行配置
        TableMapReduceUtil.initTableMapperJob(TableName.valueOf("students")
                , new Scan()
                , ReadHBaseMapper.class
                , Text.class
                , IntWritable.class
                , job
        );

        // 配置reduce任务
        // 使用HBase提供的TableMapReduceUtil工具进行配置
        TableMapReduceUtil.initTableReducerJob("clazz_num", WriteHBaseReducer.class, job);

        job.waitForCompletion(true);
        /*
        hadoop jar HBaseJavaAPI11-1.0-jar-with-dependencies.jar  Demo7MapReduceReadAndWriteHBase
         */
    }
}

Hive关联Hbase表

# Hive整合HBase
create external table students_hbase
(
id string,
name string,
age string,
gender string, 
clazz string
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties ("hbase.columns.mapping" = "
:key,
info:name,
info:age,
info:gender,
info:clazz
")
tblproperties("hbase.table.name" = "default:students");

HBaseHA

//在node1,node2中启用备用HMaster
hbase-daemon.sh start master

HBase调优

Pre-Creating Regions（预分区）

默认情况下，在创建HBase表的时候会自动创建一个region分区，当导入数据的时候，所有的HBase客户端都向这一个region写数据，直到这个region足够大了才进行切分。一种可以加快批量写入速度的方法是通过预先创建一些空的regions，这样当数据写入 HBase时，会按照region分区情况，在集群内做数据的负载均衡。
如果知道hbase数据表的key的分布情况，就可以在建表的时候对hbase进行region的预分区。这样做的好处是防止大数据量插入的热点问题，提高数据插入的效率。

//在phoenix中
CREATE TABLE IF NOT EXISTS STUDENT (
 id VARCHAR NOT NULL PRIMARY KEY, 
 name VARCHAR,
 age BIGINT, 
 gender VARCHAR ,
 clazz VARCHAR
)split on('15001006|','15001007|','15001008|') ;

//在hbase中
create 'split_table_test', 'cf', SPLITS_FILE => 'region_split_info.txt'

create 'split_table_test', 'cf', SPLITS => ['a','e','r']

Rowkey设计

原则：

唯一原则
长度原则（10-100bytes 定长）
散列原则

常用方式

反转

202133 -> 331202
202134 -> 431202
202135 -> 531202
202136 -> 631202
202137 -> 731202
202138 -> 831202
202139 -> 931202
202140 -> 041202

hash

md5 sha1

202133 -> 41DDBBCED55669818B2A40F4FED46F56
202134 -> 19D329403F02E2DA265CFC05D41FD253
202135 -> F6D06AEC4FB72A04F9CD4020BEF5E10F
202136 -> 0B512404B0411E623F64EC8981F8AE21

加上随机前缀

随机散列

第一次：202133 -> 41DDBBCED55669818B2A40F4FED46F56
第二次：202133 -> D55669818B2A40F4
第二次：202133 -> 02E2DA265CFC05D4

需求：查看某个时刻的数据

倒序、hash、但是不能用随机散列

需求：将最新的数据放到最前面

大数减小数
通常数据里有时间戳
时间戳实际上是跟Long类型非常相似一个很大的数
Long.MAX_VALUE - 值

大数：300000

202137 -> 300000 - 202137 = 97863
202138 -> 300000 - 202138 = 97862
202139 -> 300000 - 202139 = 97861
202140 -> 300000 - 202140 = 97860

加盐

会在rowkey前面加上一个随机的前缀，
优点：不需要知道rowkey的分步情况
缺点：不能再hbase中对数据进行查询和修改

CREATE TABLE IF NOT EXISTS STUDENT (
 id VARCHAR NOT NULL PRIMARY KEY, 
 name VARCHAR,
 age BIGINT, 
 gender VARCHAR ,
 clazz VARCHAR
)salt_buckets=6;

inmemory

创建表的时候，可以通过HColumnDescriptor.setInMemory(true)将表放到 RegionServer的缓存中，保证在读取的时候被cache命中。

maxversion

创建表的时候，可以通过HColumnDescriptor.setMaxVersions(int maxVersions)设置表中数据的最大版本，如果只需要保存最新版本的数据，那么可以设置 setMaxVersions(1)。

建立索引超时，查询超时

修改配置文件，hbase-site.xml

两个位置
/usr/local/soft/phoenix-4.15.0/bin
/usr/local/soft/hbase-1.4.6/conf/  所有节点

增加配置
<property>
    <name>hbase.rpc.timeout</name>
    <value>60000000</value>
</property>
<property>
    <name>hbase.client.scanner.timeout.period</name>
    <value>60000000</value>
</property>
<property>
    <name>phoenix.query.timeoutMs</name>
    <value>60000000</value>
</property>
需要重启hbase

Compact & Split

Minor Compaction：

指选取一些小的、相邻的StoreFile将他们合并成一个更大的StoreFile，在这个过程中不会处理已经Deleted或Expired的Cell。一次 Minor Compaction 的结果是更少并且更大的StoreFile。

Major Compaction：

指将所有的StoreFile合并成一个StoreFile，这个过程会清理三类没有意义的数据：被删除的数据、TTL过期数据、版本号超过设定版本号的数据。另外，一般情况下，major compaction时间会持续比较长，整个过程会消耗大量系统资源，对上层业务有比较大的影响。因此线上业务都会将关闭自动触发major compaction功能，改为手动在业务低峰期触发。

RegionSplit

region中存储的是一张表的数据，当region中的数据条数过多的时候，会直接影响查询效率。当region过大的时候，region会被拆分为两个region，HMaster会将分裂的region分配到不同的regionserver上，这样可以让请求分散到不同的RegionServer 上，已达到负载均衡 , 这也是Hbase的一个优点。

ConstantSizeRegionSplitPolicy

0.94版本前，HBase region的默认切分策略
当region中最大的store大小超过某个阈值(hbase.hregion.max.filesize=10G)之后就会触发切分，一个region等分为2个region。但是在生产线上这种切分策略却有相当大的弊端（切分策略对于大表和小表没有明显的区分）：
阈值(hbase.hregion.max.filesize)设置较大对大表比较友好，但是小表就有可能不会触发分裂，极端情况下可能就1个，形成热点，这对业务来说并不是什么好事。
如果设置较小则对小表友好，但一个大表就会在整个集群产生大量的region，这对于集群的管理、资源使用、failover来说都不是一件好事。

IncreasingToUpperBoundRegionSplitPolicy

0.94版本~2.0版本默认切分策略
总体看和ConstantSizeRegionSplitPolicy思路相同，一个region中最大的store大小大于设置阈值就会触发切分。
但是这个阈值并不像ConstantSizeRegionSplitPolicy是一个固定的值，而是会在一定条件下不断调整，调整规则和region所属表在当前regionserver上的region个数有关系.

region split阈值的计算公式是：

设regioncount：是region所属表在当前regionserver上的region的个数
阈值 = regioncount^3 * 128M * 2，当然阈值并不会无限增长，最大不超过MaxRegionFileSize（10G),当region中最大的store的大小达到该阈值的时候进行region split

例如：

第一次split阈值 = 1^3 * 256 = 256MB
第二次split阈值 = 2^3 * 256 = 2048MB
第三次split阈值 = 3^3 * 256 = 6912MB
第四次split阈值 = 4^3 * 256 = 16384MB > 10GB，因此取较小的值10GB
后面每次split的size都是10GB了

特点

相比ConstantSizeRegionSplitPolicy，可以自适应大表、小表；
在集群规模比较大的情况下，对大表的表现比较优秀
对小表不友好，小表可能产生大量的小region，分散在各regionserver上
小表达不到多次切分条件，导致每个split都很小，所以分散在各个regionServer上

SteppingSplitPolicy

2.0版本默认切分策略
相比 IncreasingToUpperBoundRegionSplitPolicy 简单了一些
region切分的阈值依然和待分裂region所属表在当前regionserver上的region个数有关系
如果region个数等于1，切分阈值为flush size 128M * 2
否则为MaxRegionFileSize。
这种切分策略对于大集群中的大表、小表会比 IncreasingToUpperBoundRegionSplitPolicy 更加友好，小表不会再产生大量的小region，而是适可而止。

KeyPrefixRegionSplitPolicy

根据rowKey的前缀对数据进行分区，这里是指定rowKey的前多少位作为前缀，比如rowKey都是16位的，指定前5位是前缀，那么前5位相同的rowKey在相同的region中。

DelimitedKeyPrefixRegionSplitPolicy

保证相同前缀的数据在同一个region中，例如rowKey的格式为：userid_eventtype_eventid，指定的delimiter为 _ ，则split的的时候会确保userid相同的数据在同一个region中。
按照分隔符进行切分，而KeyPrefixRegionSplitPolicy是按照指定位数切分。

BusyRegionSplitPolicy

按照一定的策略判断Region是不是Busy状态，如果是即进行切分
如果你的系统常常会出现热点Region，而你对性能有很高的追求，那么这种策略可能会比较适合你。它会通过拆分热点Region来缓解热点Region的压力，但是根据热点来拆分Region也会带来很多不确定性因素，因为你也不知道下一个被拆分的Region是哪个。

DisabledRegionSplitPolicy

不启用自动拆分, 需要指定手动拆分

HBase BulkLoading

优点：

如果我们一次性入库hbase巨量数据，处理速度慢不说，还特别占用Region资源，一个比较高效便捷的方法就是使用 “Bulk Loading”方法，即HBase提供的HFileOutputFormat类。
它是利用hbase的数据信息按照特定格式存储在hdfs内这一原理，直接生成这种hdfs内存储的数据格式文件，然后上传至合适位置，即完成巨量数据快速入库的办法。配合mapreduce完成，高效便捷，而且不占用region资源，增添负载。

限制：

仅适合初次数据导入，即表内数据为空，或者每次入库表内都无数据的情况。
HBase集群与Hadoop集群为同一集群，即HBase所基于的HDFS为生成HFile的MR的集群

代码

生成HFile部分

package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.RegionLocator;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.mapreduce.SimpleTotalOrderPartitioner;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Demo10BulkLoading {
    public static class BulkLoadingMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().split(",");
            String mdn = splits[0];
            String start_time = splits[1];
            // 经度
            String longitude = splits[4];
            // 维度
            String latitude = splits[5];

            String rowkey = mdn + "_" + start_time;

            KeyValue lg = new KeyValue(rowkey.getBytes(), "info".getBytes(), "lg".getBytes(), longitude.getBytes());
            KeyValue lt = new KeyValue(rowkey.getBytes(), "info".getBytes(), "lt".getBytes(), latitude.getBytes());

            context.write(new ImmutableBytesWritable(rowkey.getBytes()), lg);
            context.write(new ImmutableBytesWritable(rowkey.getBytes()), lt);

        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");


        // 创建Job实例
        Job job = Job.getInstance(conf);
        job.setJarByClass(Demo10BulkLoading.class);
        job.setJobName("Demo10BulkLoading");

        // 保证全局有序
        job.setPartitionerClass(SimpleTotalOrderPartitioner.class);

        // 设置reduce个数
        job.setNumReduceTasks(4);
        // 配置map任务
        job.setMapperClass(BulkLoadingMapper.class);

        // 配置reduce任务
        // KeyValueSortReducer 保证在每个Reduce有序
        job.setReducerClass(KeyValueSortReducer.class);

        // 输入输出路径
        FileInputFormat.addInputPath(job, new Path("/data/DIANXIN/"));
        FileOutputFormat.setOutputPath(job, new Path("/data/hfile"));

        // 创建HBase连接
        Connection conn = ConnectionFactory.createConnection(conf);
        // create 'dianxin_bulk','info'
        // 获取dianxin_bulk 表
        Table dianxin_bulk = conn.getTable(TableName.valueOf("dianxin_bulk"));
        // 获取dianxin_bulk 表 region定位器
        RegionLocator regionLocator = conn.getRegionLocator(TableName.valueOf("dianxin_bulk"));
        // 使用HFileOutputFormat2将输出的数据按照HFile的形式格式化
        HFileOutputFormat2.configureIncrementalLoad(job, dianxin_bulk, regionLocator);

        // 等到MapReduce任务执行完成
        job.waitForCompletion(true);

        // 加载HFile到 dianxin_bulk 中
        LoadIncrementalHFiles load = new LoadIncrementalHFiles(conf);
        load.doBulkLoad(new Path("/data/hfile"), conn.getAdmin(), dianxin_bulk, regionLocator);

        /**
         *  create 'dianxin_bulk','info'
         *  hadoop jar HBaseJavaAPI10-1.0-jar-with-dependencies.jar com.shujia.Demo10BulkLoading
         */
    }
}

说明

最终输出结果，无论是map还是reduce，输出部分key和value的类型必须是： < ImmutableBytesWritable, KeyValue>或者< ImmutableBytesWritable, Put>。
最终输出部分，Value类型是KeyValue 或Put，对应的Sorter分别是KeyValueSortReducer或PutSortReducer。
MR例子中HFileOutputFormat2.configureIncrementalLoad(job, dianxin_bulk, regionLocator);自动对job进行配置。SimpleTotalOrderPartitioner是需要先对key进行整体排序，然后划分到每个reduce中，保证每一个reducer中的的key最小最大值区间范围，是不会有交集的。因为入库到HBase的时候，作为一个整体的Region，key是绝对有序的。
MR例子中最后生成HFile存储在HDFS上，输出路径下的子目录是各个列族。如果对HFile进行入库HBase，相当于move HFile到HBase的Region中，HFile子目录的列族内容没有了，但不能直接使用mv命令移动，因为直接移动不能更新HBase的元数据。
HFile入库到HBase通过HBase中 LoadIncrementalHFiles的doBulkLoad方法，对生成的HFile文件入库

码农公寓

文章目录

HBase简介

HBase特点

HBase架构

Master

RegionServer

Region

Memstore&Storefile

HLog

HBase数据模型

RowKey(行键)

ColumnFamily&Qualifier（列簇和列）

TimeStamp（时间戳）

Cell(存储单元)

HBase读写流程

HBase Shell

HBaseAPI

常用java类

示例代码

HBase过滤器

作用

比较过滤器

比较运算符

常见的六大比较过滤器

BinaryComparator

BinaryPrefixComparator

NullComparator

BitComparator

RegexStringComparator

SubstringComparator

示例代码

rowKey过滤器：RowFilter

列簇过滤器：FamilyFilter

列过滤器：QualifierFilter

列值过滤器：ValueFilter

专用过滤器

单列值过滤器：SingleColumnValueFilter

列值排除过滤器：SingleColumnValueExcludeFilter

rowkey前缀过滤器：PrefixFilter

分页过滤器PageFilter

多过滤器综合查询

Phoenix

Phoenix搭建

关闭hbase集群，在master中执行

上传解压配置环境变量

将phoenix-4.15.0-HBase-1.4-server.jar复制到所有节点的hbase lib目录下

启动hbase ， 在master中执行

配置环境变量

Phoenix使用

连接sqlline

常用命令

phoenix映射

视图映射

表映射

Phoenix二级索引

开启索引支持

创建索引

全局索引

本地索引

全局索引与本地索引区别

覆盖索引

PhoenixJDBC

HBase SQL与Hive SQL执行流程

HBase的MapReduce过程

Hive关联Hbase表

HBaseHA

HBase调优

Pre-Creating Regions（预分区）

Rowkey设计

原则：

常用方式

反转

hash

md5 sha1

加上随机前缀

随机散列

需求：查看某个时刻的数据

需求：将最新的数据放到最前面

加盐

启动hbase ，在master中执行