• Hbase

    1、HBase集群安装部署

        hbase安装包

    （1）、解压文件

    cd /kkb/soft/
    tar -xzvf hbase-1.2.0-cdh5.14.2.tar.gz -C /kkb/install/

    （2）、修改HBase配置文件

    cd /kkb/install/hbase-1.2.0-cdh5.14.2/conf/
    vim hbase-env.sh

     添加java地址

    export JAVA_HOME=/opt/install/jdk1.8.0_141
    export HBASE_MANAGES_ZK=false 

     （3）、修改 hbase-site.xml

    vim hbase-site.xml


        
            hbase.rootdir
            hdfs://node01:8020/hbase  
        
        
            hbase.cluster.distributed
            true
        
        
        
            hbase.master.port
            16000
        
        
            hbase.zookeeper.quorum
            node01,node02,node03
        
       
        
            hbase.zookeeper.property.clientPort
            2181
        
        
            hbase.zookeeper.property.dataDir
            /kkb/install/zookeeper-3.4.5-cdh5.14.2/zkdatas
        
       
        
            zookeeper.znode.parent
            /hbase
        
     

    （4）、egionservers修改

    [hadoop@node01 conf]$ vim regionservers

    • 指定HBase集群的从节点；原内容清空，添加如下三行

    node01
node02
node03

    （5）、back-masters

    • 创建back-masters配置文件，里边包含备份HMaster节点的主机名，每个机器独占一行，实现HMaster的高可用

    • vim backup-masters

    • 将node02作为备份的HMaster节点，问价内容如下

    node02 

    （6）、分发安装包

     cd /kkb/install
    scp -r hbase-1.2.0-cdh5.14.2/ node02:$PWD

    scp -r hbase-1.2.0-cdh5.14.2/ node03:$PWD

    （7）、建立软连接

    • 注意：三台机器均做如下操作

    • 因为HBase集群需要读取hadoop的core-site.xml、hdfs-site.xml的配置文件信息，所以我们==三台机器==都要执行以下命令，在相应的目录创建这两个配置文件的软连接

    ln -s /kkb/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop/core-site.xml  /kkb/install/hbase-1.2.0-cdh5.14.2/conf/core-site.xml

    ln -s /kkb/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop/hdfs-site.xml  /kkb/install/hbase-1.2.0-cdh5.14.2/conf/hdfs-site.xml

    （7）、添加HBase环境变量

    • 注意：三台机器均执行以下命令，添加环境变量

    vim /etc/profile

    export HBASE_HOME=/kkb/install/hbase-1.2.0-cdh5.14.2
    export PATH=$PATH:$HBASE_HOME/bin

    重新编译/etc/profile，让环境变量生效

     source /etc/profile

    （8）、Base的启动与停止

    • 需要提前启动HDFS及ZooKeeper集群

    • 第一台机器==node01==（HBase主节点）执行以下命令，启动HBase集群

     start-hbase.sh

    • 启动完后，jps查看HBase相关进程

      node01、node02上有进程HMaster、HRegionServer

      node03上有进程HRegionServer

    • 警告提示：HBase启动的时候会产生一个警告，这是因为jdk7与jdk8的问题导致的，如果linux服务器安装jdk8就会产生这样的一个警告  

    查看当前所有节点Jps状态

    2、hbase基础学习

    2.1、rowkey行键

    • table的主键，table中的记录==按照rowkey 的字典序进行排序==

    • Row key行键可以是任意字符串(最大长度是 64KB，实际应用中长度一般为 10-100bytes)

    2.2 Column Family列族

    • 列族或列簇

    • HBase表中的每个列，都归属与某个列族

    • 列族是表的schema的一部分(而列不是)，即建表时至少指定一个列族

    • 比如创建一张表，名为user，有两个列族，分别是info和data，建表语句create 'user', 'info', 'data'

    2.3 Column列

    • 列肯定是表的某一列族下的一个列，用列族名:列名表示，如info列族下的name列，表示为info:name

    • 属于某一个ColumnFamily,类似于我们mysql当中创建的具体的列

    2.4 cell单元格

    • 指定row key行键、列族、列，可以确定的一个cell单元格

    • cell中的数据是没有类型的，全部是以字节数组进行存储

    2.5 Timestamp时间戳

    • 可以对表中的Cell多次赋值，每次赋值操作时的时间戳timestamp，可看成Cell值的版本号version number

    • 即一个Cell可以有多个版本的值

    Client客户端

    • Client是操作HBase集群的入口

      • 对于管理类的操作，如表的增、删、改操纵，Client通过RPC与HMaster通信完成

      • 对于表数据的读写操作，Client通过RPC与RegionServer交互，读写数据

    • Client类型：

      • HBase shell

      • Java编程接口

      • Thrift、Avro、Rest等等

     ZooKeeper集群

    • 作用

      • 实现了HMaster的高可用，多HMaster间进行主备选举

      • 保存了HBase的元数据信息meta表，提供了HBase表中region的寻址入口的线索数据

      • 对HMaster和HRegionServer实现了监控

     HMaster

    • HBase集群也是主从架构，HMaster是主的角色，是老大

    • 主要负责Table表和Region的相关管理工作：

    • 关于Table

      • 管理Client对Table的增删改的操作

      • 关于Region

        • 在Region分裂后，负责新Region分配到指定的HRegionServer上

        • 管理HRegionServer间的负载均衡，迁移region分布

        • 当HRegionServer宕机后，负责其上的region的迁移

    HRegionServer

    • HBase集群中从的角色，是小弟

    • 作用

      • 响应客户端的读写数据请求

      • 负责管理一系列的Region

      • 切分在运行过程中变大的region

     Region

    • HBase集群中分布式存储的最小单元

    • 一个Region对应一个Table表的部分数据

    基础命令操作 

    进入hbase

    hbase shell 

    创建user表，以及info，data列族 

    create 'user', 'info', 'data' 

    向user表中插入信息，row key为rk0001，列族info中添加名为name的列，值为zhangsan 


    HBase(main):011:0> put 'user', 'rk0001', 'info:name', 'zhangsan' 

    根据rowkey查询该行所有列族的数据

    get 'user', 'rk0001' 

     获取user表中row key为rk0001，info列族的name、age列的信息

    get 'user', 'rk0001', 'info:name', 'info:age'

    过滤值， 获取user表中row key为rk0001，cell的值为zhangsan的信息

     get 'user', 'rk0001', {FILTER => "ValueFilter(=, 'binary:zhangsan')"}

    获取user表中row key为rk0001，列标示符中含有a的信息  

    get 'user', 'rk0001', {FILTER => "QualifierFilter(=,'substring:a')"} 

     

     查看所有的表

    list

    查看表中所有数据

    scan 'user' 

    查看user表中，一个列族的所有信息 

     scan 'user', {COLUMNS => 'info'} 

     查询info:name列、data:pic列的数据

    scan 'user', {COLUMNS => ['info:name', 'data:pic']}

    将user表的f1列族版本数改为5

    alter 'user', NAME => 'info', VERSIONS => 5 

    删除user表row key为rk0001，列标示符为info:name的数据  

    delete 'user', 'rk0001', 'info:name' 

    删除user表row key为rk0001，列标示符为info:name，timestamp为1392383705316的数据  

    delete 'user', 'rk0001', 'info:name', 1392383705316 

    清空表数据

    truncate 'user'

     删除表

    • 首先需要先让该表为disable状态，使用命令：

    disable 'user'

    • 然后使用drop命令删除这个表

     drop 'user'

    (注意：如果直接drop表，会报错：Drop the named table. Table must first be disabled)

    查看所有的表

    list 

    统计表中有多少个行键（rowkey） 

    count 'user'

    javaApi操作 

    pom.xml

    1. <dependencies>
    2.     <dependency>
    3.         <groupId>org.apache.hadoopgroupId>
    4.         <artifactId>hadoop-clientartifactId>
    5.         <version>2.6.0-mr1-cdh5.14.2version>
    6.     dependency>
    7.     <dependency>
    8.         <groupId>org.apache.hbasegroupId>
    9.         <artifactId>hbase-clientartifactId>
    10.         <version>1.2.0-cdh5.14.2version>
    11.     dependency>
    12.     <dependency>
    13.         <groupId>org.apache.hbasegroupId>
    14.         <artifactId>hbase-serverartifactId>
    15.         <version>1.2.0-cdh5.14.2version>
    16.     dependency>
    17.     <dependency>
    18.         <groupId>junitgroupId>
    19.         <artifactId>junitartifactId>
    20.         <version>4.12version>
    21.         <scope>testscope>
    22.     dependency>
    23.     <dependency>
    24.         <groupId>org.testnggroupId>
    25.         <artifactId>testngartifactId>
    26.         <version>6.14.3version>
    27.         <scope>testscope>
    28.     dependency>
    29.     <dependency>
    30.         <groupId>org.junit.jupitergroupId>
    31.         <artifactId>junit-jupiterartifactId>
    32.         <version>RELEASEversion>
    33.         <scope>compilescope>
    34.     dependency>
    35.     <dependency>
    36.         <groupId>org.junit.jupitergroupId>
    37.         <artifactId>junit-jupiterartifactId>
    38.         <version>RELEASEversion>
    39.         <scope>compilescope>
    40.     dependency>
    41.     <dependency>
    42.         <groupId>junitgroupId>
    43.         <artifactId>junitartifactId>
    44.         <version>4.13.1version>
    45.         <scope>compilescope>
    46.     dependency>
    47.  
    48. dependencies>

    1、创建表

    1.     @Test
    2.     public void createTable() throws IOException {
    3.         Configuration configuration = HBaseConfiguration.create();
    4.         //连接HBase集群不需要指定HBase主节点的ip地址和端口号
    5.         configuration.set("hbase.zookeeper.quorum","node01:2181,node02:2181,node03:2181");
    6.         //创建连接对象
    7.         Connection connection = ConnectionFactory.createConnection(configuration);
    8.         //获取连接对象，创建一张表
    9.         //获取管理员对象，来对手数据库进行DDL的操作
    10.         Admin admin = connection.getAdmin();
    11.         //指定我们的表名
    12.         TableName myuser = TableName.valueOf("myuser");
    13.         HTableDescriptor hTableDescriptor = new HTableDescriptor(myuser);
    14.         //指定两个列族
    15.         HColumnDescriptor f1 = new HColumnDescriptor("f1");
    16.         HColumnDescriptor f2 = new HColumnDescriptor("f2");
    17.         hTableDescriptor.addFamily(f1);
    18.         hTableDescriptor.addFamily(f2);
    19.  
    20.         admin.createTable(hTableDescriptor);
    21.         admin.close();
    22.         connection.close();
    23.     }

    2、添加数据

    1.   private Connection connection;
    2.     private final String TABLE_NAME = "myuser";
    3.     private Table table;
    4.  
    5.     @Before
    6.     public void initTable() throws IOException {
    7.         System.out.println("1111111111");
    8.         Configuration configuration = HBaseConfiguration.create();
    9.         configuration.set("hbase.zookeeper.quorum", "node01:2181,node02:2181");
    10.         connection = ConnectionFactory.createConnection(configuration);
    11.         table = connection.getTable(TableName.valueOf(TABLE_NAME));
    12.     }
    13.  
    14.     @After
    15.     public void close() throws IOException {
    16.         System.out.println("222222");
    17.         table.close();
    18.         connection.close();
    19.     }
    20.  
    21.     /**
    22.      * 向myuser表当中添加数据
    23.      */
    24.     @Test
    25.     public void addData() throws IOException {
    26.         //获取表
    27.         //Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
    28.         Put put = new Put("0001".getBytes());//创建put对象，并指定rowkey值
    29.         put.addColumn("f2".getBytes(), "name".getBytes(), "zhangsan".getBytes());
    30.         put.addColumn("f2".getBytes(), "age".getBytes(), Bytes.toBytes(18));
    31.         put.addColumn("f2".getBytes(), "id".getBytes(), Bytes.toBytes(25));
    32.         put.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("地球人"));
    33.         System.out.println("000000000000000");
    34.         table.put(put);
    35.         table.close();
    36.     }

    3、批量插入数据 

    1.    /**
    2.      * hbase的批量插入数据
    3.      */
    4.     @Test
    5.     public void batchInsert() throws IOException {
    6.         //创建put对象，并指定rowkey
    7.         Put put = new Put("0002".getBytes());
    8.         //f1
    9.         put.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(1));
    10.         put.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("曹操"));
    11.         put.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(30));
    12.         //f2
    13.         put.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    14.         put.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("沛国谯县"));
    15.         put.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("16888888888"));
    16.         put.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("helloworld"));
    17.  
    18.         Put put2 = new Put("0003".getBytes());
    19.         put2.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(2));
    20.         put2.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("刘备"));
    21.         put2.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(32));
    22.         put2.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    23.         put2.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("幽州涿郡涿县"));
    24.         put2.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("17888888888"));
    25.         put2.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("talk is cheap , show me the code"));
    26.  
    27.         Put put3 = new Put("0004".getBytes());
    28.         put3.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(3));
    29.         put3.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("孙权"));
    30.         put3.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(35));
    31.         put3.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    32.         put3.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("下邳"));
    33.         put3.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("12888888888"));
    34.         put3.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("what are you 弄啥嘞！"));
    35.  
    36.         Put put4 = new Put("0005".getBytes());
    37.         put4.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(4));
    38.         put4.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("诸葛亮"));
    39.         put4.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(28));
    40.         put4.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    41.         put4.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("四川隆中"));
    42.         put4.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("14888888888"));
    43.         put4.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("出师表你背了嘛"));
    44.  
    45.         Put put5 = new Put("0006".getBytes());
    46.         put5.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(5));
    47.         put5.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("司马懿"));
    48.         put5.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(27));
    49.         put5.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    50.         put5.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("哪里人有待考究"));
    51.         put5.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15888888888"));
    52.         put5.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("跟诸葛亮死掐"));
    53.  
    54.  
    55.         Put put6 = new Put("0007".getBytes());
    56.         put6.addColumn("f1".getBytes(), "id".getBytes(), Bytes.toBytes(5));
    57.         put6.addColumn("f1".getBytes(), "name".getBytes(), Bytes.toBytes("xiaobubu—吕布"));
    58.         put6.addColumn("f1".getBytes(), "age".getBytes(), Bytes.toBytes(28));
    59.         put6.addColumn("f2".getBytes(), "sex".getBytes(), Bytes.toBytes("1"));
    60.         put6.addColumn("f2".getBytes(), "address".getBytes(), Bytes.toBytes("内蒙人"));
    61.         put6.addColumn("f2".getBytes(), "phone".getBytes(), Bytes.toBytes("15788888888"));
    62.         put6.addColumn("f2".getBytes(), "say".getBytes(), Bytes.toBytes("貂蝉去哪了"));
    63.  
    64.         List listPut = new ArrayList();
    65.         listPut.add(put);
    66.         listPut.add(put2);
    67.         listPut.add(put3);
    68.         listPut.add(put4);
    69.         listPut.add(put5);
    70.         listPut.add(put6);
    71.  
    72.         table.put(listPut);
    73.     }

    4、行键rowkey的查询

    1. /**
    2.      * 查询rowkey为0003的人
    3.      * get -> Result
    4.      */
    5.     @Test
    6.     public void getData() throws IOException {
    7.         //Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
    8.         //通过get对象，指定rowkey
    9.         Get get = new Get(Bytes.toBytes("0007"));
    10.  
    11.         get.addFamily("f1".getBytes());//限制只查询f1列族下面所有列的值
    12.         //查询f2  列族 phone  这个字段
    13.         get.addColumn("f2".getBytes(), "address".getBytes());
    14.         //通过get查询，返回一个result对象，所有的字段的数据都是封装在result里面了
    15.  
    16.         Result result = table.get(get);
    17.         List cells = result.listCells();  //获取一条数据所有的cell，所有数据值都是在cell里面 的
    18.  
    19.         if (cells != null) {
    20.             for (Cell cell : cells) {
    21.                 byte[] family_name = CellUtil.cloneFamily(cell);//获取列族名
    22.                 byte[] column_name = CellUtil.cloneQualifier(cell);//获取列名
    23.                 byte[] rowkey = CellUtil.cloneRow(cell);//获取rowkey
    24.                 byte[] cell_value = CellUtil.cloneValue(cell);//获取cell值
    25.                 //需要判断字段的数据类型，使用对应的转换的方法，才能够获取到值
    26.                 if ("age".equals(Bytes.toString(column_name)) || "id".equals(Bytes.toString(column_name))) {
    27.                     System.out.println(Bytes.toString(family_name));
    28.                     System.out.println(Bytes.toString(column_name));
    29.                     System.out.println(Bytes.toString(rowkey));
    30.                     System.out.println(Bytes.toInt(cell_value));
    31.                 } else {
    32.                     System.out.println(Bytes.toString(family_name));
    33.                     System.out.println(Bytes.toString(column_name));
    34.                     System.out.println(Bytes.toString(rowkey));
    35.                     System.out.println(Bytes.toString(cell_value));
    36.                 }
    37.             }
    38.             //table.close();
    39.         }
    40.     }
    5、不知道rowkey的具体值，我想查询rowkey范围值是0003  到0006
    1.  @Test
    2.     public void scanData() throws IOException {
    3.         //获取table
    4.         //Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
    5.         Scan scan = new Scan();//没有指定startRow以及stopRow  全表扫描
    6.         //只扫描f1列族
    7.         scan.addFamily("f1".getBytes());
    8.         //扫描 f2列族 phone  这个字段
    9.         scan.addColumn("f2".getBytes(), "phone".getBytes());
    10.         scan.setStartRow("0003".getBytes());
    11.         scan.setStopRow("0007".getBytes());
    12.         //通过getScanner查询获取到了表里面所有的数据，是多条数据
    13.         ResultScanner scanner = table.getScanner(scan);
    14.         //遍历ResultScanner 得到每一条数据，每一条数据都是封装在result对象里面了
    15.         for (Result result : scanner) {
    16.             List cells = result.listCells();
    17.             for (Cell cell : cells) {
    18.                 byte[] family_name = CellUtil.cloneFamily(cell);
    19.                 byte[] qualifier_name = CellUtil.cloneQualifier(cell);
    20.                 byte[] rowkey = CellUtil.cloneRow(cell);
    21.                 byte[] value = CellUtil.cloneValue(cell);
    22.                 //判断id和age字段，这两个字段是整形值
    23.                 if ("age".equals(Bytes.toString(qualifier_name)) || "id".equals(Bytes.toString(qualifier_name))) {
    24.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toInt(value));
    25.                 } else {
    26.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toString(value));
    27.                 }
    28.             }
    29.         }
    30.         //table.close();
    31.     }

     6、查询所有rowkey比0003小的数据

    1.   @Test
    2.     public void rowFilter() throws IOException {
    3.         //Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
    4.         Scan scan = new Scan();
    5.         //获取我们比较对象
    6.         BinaryComparator binaryComparator = new BinaryComparator("0003".getBytes());
    7.         /***
    8.          * rowFilter需要加上两个参数
    9.          * 第一个参数就是我们的比较规则
    10.          * 第二个参数就是我们的比较对象
    11.          */
    12.         RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryComparator);
    13.         //为我们的scan对象设置过滤器
    14.         scan.setFilter(rowFilter);
    15.         SubstringComparator substringComparator = new SubstringComparator("f2");
    16.         //通过familyfilter来设置列族的过滤器
    17. //        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
    18. //        scan.se(familyFilter);
    19.         ResultScanner scanner = table.getScanner(scan);
    20.         for (Result result : scanner) {
    21.             List<Cell> cells = result.listCells();
    22.             for (Cell cell : cells) {
    23.                 byte[] family_name = CellUtil.cloneFamily(cell);
    24.                 byte[] qualifier_name = CellUtil.cloneQualifier(cell);
    25.                 byte[] rowkey = CellUtil.cloneRow(cell);
    26.                 byte[] value = CellUtil.cloneValue(cell);
    27.                 //判断id和age字段，这两个字段是整形值
    28.                 if ("age".equals(Bytes.toString(qualifier_name)) || "id".equals(Bytes.toString(qualifier_name))) {
    29.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toInt(value));
    30.                 } else {
    31.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toString(value));
    32.                 }
    33.             }
    34.         }
    35.     }

    7、列族过滤器FamilyFilter

    • 查询列族名包含f2的所有列族下面的数据

    1.  /**
    2.      * 通过familyFilter来实现列族的过滤
    3.      * 需要过滤，列族名包含f2
    4.      * f1  f2   hello   world
    5.      */
    6.     @Test
    7.     public void familyFilter() throws IOException {
    8.         Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
    9.         Scan scan = new Scan();
    10.         SubstringComparator substringComparator = new SubstringComparator("f2");
    11.         //通过familyfilter来设置列族的过滤器
    12.         FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
    13.         scan.setFilter(familyFilter);
    14.         ResultScanner scanner = table.getScanner(scan);
    15.         for (Result result : scanner) {
    16.             List cells = result.listCells();
    17.             for (Cell cell : cells) {
    18.                 byte[] family_name = CellUtil.cloneFamily(cell);
    19.                 byte[] qualifier_name = CellUtil.cloneQualifier(cell);
    20.                 byte[] rowkey = CellUtil.cloneRow(cell);
    21.                 byte[] value = CellUtil.cloneValue(cell);
    22.                 //判断id和age字段，这两个字段是整形值
    23.                 if ("age".equals(Bytes.toString(qualifier_name)) || "id".equals(Bytes.toString(qualifier_name))) {
    24.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toInt(value));
    25.                 } else {
    26.                     System.out.println("数据的rowkey为" + Bytes.toString(rowkey) + "======数据的列族为" + Bytes.toString(family_name) + "======数据的列名为" + Bytes.toString(qualifier_name) + "==========数据的值为" + Bytes.toString(value));
    27.                 }
    28.             }
    29.         }
    30.     }
    1. 数据的rowkey为0001======数据的列族为f2======数据的列名为address==========数据的值为地球人
    2. 数据的rowkey为0001======数据的列族为f2======数据的列名为age==========数据的值为18
    3. 数据的rowkey为0001======数据的列族为f2======数据的列名为id==========数据的值为25
    4. 数据的rowkey为0001======数据的列族为f2======数据的列名为name==========数据的值为zhangsan
    5. 数据的rowkey为0002======数据的列族为f2======数据的列名为address==========数据的值为沛国谯县
    6. 数据的rowkey为0002======数据的列族为f2======数据的列名为phone==========数据的值为16888888888
    7. 数据的rowkey为0002======数据的列族为f2======数据的列名为say==========数据的值为helloworld
    8. 数据的rowkey为0002======数据的列族为f2======数据的列名为sex==========数据的值为1
    9. 数据的rowkey为0003======数据的列族为f2======数据的列名为address==========数据的值为幽州涿郡涿县
    10. 数据的rowkey为0003======数据的列族为f2======数据的列名为phone==========数据的值为17888888888
    11. 数据的rowkey为0003======数据的列族为f2======数据的列名为say==========数据的值为talk is cheap , show me the code
    12. 数据的rowkey为0003======数据的列族为f2======数据的列名为sex==========数据的值为1
    13. 数据的rowkey为0004======数据的列族为f2======数据的列名为address==========数据的值为下邳
    14. 数据的rowkey为0004======数据的列族为f2======数据的列名为phone==========数据的值为12888888888
    15. 数据的rowkey为0004======数据的列族为f2======数据的列名为say==========数据的值为what are you 弄啥嘞！
    16. 数据的rowkey为0004======数据的列族为f2======数据的列名为sex==========数据的值为1
    17. 数据的rowkey为0005======数据的列族为f2======数据的列名为address==========数据的值为四川隆中
    18. 数据的rowkey为0005======数据的列族为f2======数据的列名为phone==========数据的值为14888888888
    19. 数据的rowkey为0005======数据的列族为f2======数据的列名为say==========数据的值为出师表你背了嘛
    20. 数据的rowkey为0005======数据的列族为f2======数据的列名为sex==========数据的值为1
    21. 数据的rowkey为0006======数据的列族为f2======数据的列名为address==========数据的值为哪里人有待考究
    22. 数据的rowkey为0006======数据的列族为f2======数据的列名为phone==========数据的值为15888888888
    23. 数据的rowkey为0006======数据的列族为f2======数据的列名为say==========数据的值为跟诸葛亮死掐
    24. 数据的rowkey为0006======数据的列族为f2======数据的列名为sex==========数据的值为1
    25. 数据的rowkey为0007======数据的列族为f2======数据的列名为address==========数据的值为内蒙人
    26. 数据的rowkey为0007======数据的列族为f2======数据的列名为phone==========数据的值为15788888888
    27. 数据的rowkey为0007======数据的列族为f2======数据的列名为say==========数据的值为貂蝉去哪了
    28. 数据的rowkey为0007======数据的列族为f2======数据的列名为sex==========数据的值为1

    8、列过滤器QualifierFilter

    1. /**
    2.      * 列名过滤器 只查询包含name列的值
    3.      */
    4.     @Test
    5.     public void  qualifierFilter() throws IOException {
    6.         Scan scan = new Scan();
    7.         SubstringComparator substringComparator = new SubstringComparator("name");
    8.         //定义列名过滤器，只查询列名包含name的列
    9.         QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
    10.         scan.setFilter(qualifierFilter);
    11.         ResultScanner scanner = table.getScanner(scan);
    12.         printResult(scanner);
    13.     }

    9、列值过滤器ValueFilter

    • 查询所有列当中包含8的数据

    1.  /**
    2.      * 查询哪些字段值  包含数字8
    3.      */
    4.     @Test
    5.     public void contains8() throws IOException {
    6.         Scan scan = new Scan();
    7.         SubstringComparator substringComparator = new SubstringComparator("8");
    8.         //列值过滤器，过滤列值当中包含数字8的所有的列
    9.         ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, substringComparator);
    10.         scan.setFilter(valueFilter);
    11.         ResultScanner scanner = table.getScanner(scan);
    12.         printResult(scanner);
    13.     }

    10、专用过滤器使用

    1、单列值过滤器 SingleColumnValueFilter

    • SingleColumnValueFilter会返回满足条件的cell。所在行的所有cell的值

    • 查询名字为刘备的数据

    1. /**
    2.      * select  *  from  myuser where name  = '刘备'
    3.      * 会返回我们符合条件数据的所有的字段
    4.      *
    5.      * SingleColumnValueExcludeFilter  列值排除过滤器
    6.      *  select  *  from  myuser where name  ！= '刘备'
    7.      */
    8.     @Test
    9.     public void singleColumnValueFilter() throws IOException {
    10.         //查询 f1  列族 name  列  值为刘备的数据
    11.         Scan scan = new Scan();
    12.         //单列值过滤器，过滤  f1 列族  name  列  值为刘备的数据
    13.         SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("f1".getBytes(), "name".getBytes(), CompareFilter.CompareOp.EQUAL, "刘备".getBytes());
    14.         scan.setFilter(singleColumnValueFilter);
    15.         ResultScanner scanner = table.getScanner(scan);
    16.         printResult(scanner);
    17.     }

    11、列值排除过滤器SingleColumnValueExcludeFilter

    • 与SingleColumnValueFilter相反

      • 如果指定列的值符合filter条件，则会排除掉row中指定的列，其他的列全部返回

      • 如果列不存在或不符合filter条件，则不返回row中的列

    11、rowkey前缀过滤器PrefixFilter

    • 查询以00开头的所有前缀的rowkey

    1. 	/**
    2.      * 查询rowkey前缀以  00开头的所有的数据
    3.      */
    4.     @Test
    5.     public  void  prefixFilter() throws IOException {
    6.         Scan scan = new Scan();
    7.         //过滤rowkey以  00开头的数据
    8.         PrefixFilter prefixFilter = new PrefixFilter("00".getBytes());
    9.         scan.setFilter(prefixFilter);
    10.         ResultScanner scanner = table.getScanner(scan);
    11.         printlReult(scanner);
    12.     }

    12、分页过滤器PageFilter

    • 通过pageFilter实现分页过滤器

    1. /**
    2.      * HBase当中的分页
    3.      */
    4.     @Test
    5.     public void hbasePageFilter() throws IOException {
    6.         int pageNum= 3;
    7.         int pageSize = 2;
    8.         Scan scan = new Scan();
    9.         if(pageNum == 1 ){
    10.             //获取第一页的数据
    11.             //scan.setMaxResultSize(pageSize);
    12.             scan.setStartRow("".getBytes());
    13.             //使用分页过滤器来实现数据的分页
    14.             PageFilter filter = new PageFilter(pageSize);
    15.             scan.setFilter(filter);
    16.             ResultScanner scanner = table.getScanner(scan);
    17.             printResult(scanner);
    18.         }else{
    19.             String  startRow = "";
    20.             //扫描数据的调试 扫描五条数据
    21.             int scanDatas = (pageNum - 1) * pageSize + 1;
    22.             //scan.setMaxResultSize(scanDatas);//设置一步往前扫描多少条数据
    23.             PageFilter filter = new PageFilter(scanDatas);
    24.             scan.setFilter(filter);
    25.             ResultScanner scanner = table.getScanner(scan);
    26.             for (Result result : scanner) {
    27.                 byte[] row = result.getRow();//获取rowkey
    28.                 //最后一次startRow的值就是0005
    29.                 startRow= Bytes.toString(row);//循环遍历我们多有获取到的数据的rowkey
    30.                 //最后一条数据的rowkey就是我们需要的起始的rowkey
    31.             }
    32.             //获取第三页的数据
    33.             scan.setStartRow(startRow.getBytes());
    34.             //scan.setMaxResultSize(pageSize);//设置我们扫描多少条数据
    35.             PageFilter filter1 = new PageFilter(pageSize);
    36.             scan.setFilter(filter1);
    37. //            scan.setAllowPartialResults(true);
    38. //            scan.setBatch(100);
    39.             ResultScanner scanner1 = table.getScanner(scan);
    40.             printResult(scanner1);
    41.         }
    42.     }

     分页过滤器二

    1. @Test
    2.     public void hbasePageFilter2() throws IOException {
    3.  
    4.         Scan scan = new Scan();
    5.         int pageNum = 2;
    6.         int pageSize = 2;
    7.         String startRow = "000" + (pageNum - 1) * pageSize + 1;
    8.  
    9.         //获取第三页的数据
    10.         scan.setStartRow(startRow.getBytes());
    11.         //scan.setMaxResultSize(pageSize);//设置我们扫描多少条数据
    12.         PageFilter filter1 = new PageFilter(pageSize);
    13.         scan.setFilter(filter1);
    14. //            scan.setAllowPartialResults(true);
    15. //            scan.setBatch(100);
    16.         ResultScanner scanner1 = table.getScanner(scan);
    17.         printResult(scanner1);
    18.  
    19.     }

    13、多过滤器综合查询FilterList

    • 需求：使用SingleColumnValueFilter查询f1列族，name为刘备的数据，并且同时满足rowkey的前缀以00开头的数据（PrefixFilter）

    1.    @Test
    2.     public void filterList() throws IOException {
    3.         Scan scan = new Scan();
    4.         SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("f1".getBytes(), "name".getBytes(), CompareFilter.CompareOp.EQUAL, "刘备".getBytes());
    5.         PrefixFilter prefixFilter = new PrefixFilter("00".getBytes());
    6.         FilterList filterList = new FilterList();
    7.         filterList.addFilter(singleColumnValueFilter);
    8.         filterList.addFilter(prefixFilter);
    9.         scan.setFilter(filterList);
    10.         ResultScanner scanner = table.getScanner(scan);
    11.         printResult(scanner);
    12.     }

    14、根据rowkey删除数据

    • 删除rowkey为003的数据

    1. /**
    2.      * 删除数据
    3.      */
    4.     @Test
    5.     public  void  deleteData() throws IOException {
    6.         Delete delete = new Delete("0003".getBytes());
    7.         table.delete(delete);
    8.     }

    15、删除表

    1. /**
    2.      * 删除表
    3.      */
    4.     @Test
    5.     public void deleteTable() throws IOException {
    6.         //获取管理员对象，用于表的删除
    7.         Admin admin = connection.getAdmin();
    8.         //删除一张表之前，需要先禁用表
    9.         admin.disableTable(TableName.valueOf(TABLE_NAME));
    10.         admin.deleteTable(TableName.valueOf(TABLE_NAME));
    11.     }

    HBase读写流程

    • 第一部分: HLog, WAL, 存储写入数据的指令, 而不是数据, HLog存储在hdfs之上

    • 第二部分: BlockCache

    • 第三部分: Region

      • 多个store模块, store模块和列族有关系

        • memoryStore

        • 多个StoreFile(HFile)

    • RegionServer --> n个Region

    • RegionServer -> 1HLog

    • RegionServer -> 1 BlockCache

    • Region -> n个store模块

    • store -> 1memorystroe(128) + n个storeFile

    1. Cli连接zk, 获取一张特殊表, hbase/meta表, 获取这张表对应的region, 存储在哪一个regionserver上, 然后将对应的信息缓存在客户端

    2. 客户端连接对应的regionserver, 读取对应的meta的信息(存储了hbase的元数据信息, 有哪些hbase表, 表对应的region在哪里, 以及每一个region管理的数据范围), hbase/meta表存储的是hbase的元数据信息 scan 'hbase:meta'

    存储到blockcache: 为了提高后期查询的效率 (为了提高读的效率)

    memorystore: 是为了提高写入的效率, 如果直接把文件写入到磁盘(阻塞) io效率差

    • 如果memorystore中的数据超过128M, memorystore中的数据会进入到一个flush队列, 书写到StoreFile

    • 刷写完成之后, 删除HLog中的历史数据

    1、hbase读取数据流程 

    说明：HBase集群，只有一张meta表，此表只有一个region，该region数据保存在一个HRegionServer上

    • 1、客户端首先与zk进行连接；

      • 从zk找到meta表的region位置，即meta表的数据存储在某一HRegionServer上；

      • 客户端与此HRegionServer建立连接，然后读取meta表中的数据；meta表中存储了所有用户表的region信息，我们可以通过scan 'hbase:meta'来查看meta表信息

    • 2、根据要查询的namespace、表名和rowkey信息。找到写入数据对应的region信息

    • 3、找到这个region对应的regionServer，然后发送请求

    • 4、查找并定位到对应的region

    • 5、先从memstore查找数据，如果没有，再从BlockCache上读取

      • HBase上Regionserver的内存分为两个部分

        • 一部分作为Memstore，主要用来写；

        • 另外一部分作为BlockCache，主要用于读数据；

    • 6、如果BlockCache中也没有找到，再到StoreFile上进行读取

      • 从storeFile中读取到数据之后，不是直接把结果数据返回给客户端，而是把数据先写入到BlockCache中，目的是为了加快后续的查询；然后再返回结果给客户端。

    HBase写数据流程

    • 1、客户端首先从zk找到meta表的region位置，然后读取meta表中的数据，meta表中存储了用户表的region信息

    • 2、根据namespace、表名和rowkey信息。找到写入数据对应的region信息

    • 3、找到这个region对应的regionServer，然后发送请求

    • 4、把数据分别写到HLog（write ahead log）和memstore各一份

    • 5、memstore达到阈值后把数据刷到磁盘，生成storeFile文件

    • 6、删除HLog中的历史数据

    补充：
    HLog（write ahead log）：
        也称为WAL意为Write ahead log，类似mysql中的binlog,用来做灾难恢复时用，HLog记录数据的所有变更,一旦数据修改，就可以从log中进行恢复。 

     MemStore的flush

    1、Flush触发条件

    1、 memstore级别限制

    • 当Region中任意一个MemStore的大小达到了上限（hbase.hregion.memstore.flush.size，默认128MB），会触发Memstore刷新。

    
    hbase.hregion.memstore.flush.size
    134217728

    2、region级别限制

    • 当Region中所有Memstore的大小总和达到了上限（hbase.hregion.memstore.block.multiplier * hbase.hregion.memstore.flush.size，默认 2* 128M = 256M），会触发memstore刷新。

    
    hbase.hregion.memstore.flush.size
    134217728


    hbase.hregion.memstore.block.multiplier
    4
  

    3、Region Server级别限制

    • 当一个Region Server中所有Memstore的大小总和超过低水位阈值hbase.regionserver.global.memstore.size.lower.limit*hbase.regionserver.global.memstore.size（前者默认值0.95），RegionServer开始强制flush；

    • 先Flush Memstore最大的Region，再执行次大的，依次执行；

    • 如写入速度大于flush写出的速度，导致总MemStore大小超过高水位阈值hbase.regionserver.global.memstore.size（默认为JVM内存的40%），此时RegionServer会阻塞更新并强制执行flush，直到总MemStore大小低于低水位阈值

    
    hbase.regionserver.global.memstore.size.lower.limit
    0.95


    hbase.regionserver.global.memstore.size
    0.4

    4、 HLog数量上限

    • 当一个Region Server中HLog数量达到上限（可通过参数hbase.regionserver.maxlogs配置）时，系统会选取最早的一个 HLog对应的一个或多个Region进行flush

    5、定期刷新Memstore

    • 默认周期为1小时，确保Memstore不会长时间没有持久化。为避免所有的MemStore在同一时间都进行flush导致的问题，定期的flush操作有20000左右的随机延时。

    6、手动flush

    • 用户可以通过shell命令flush ‘tablename’或者flush ‘region name’分别对一个表或者一个Region进行flush。

    flush的流程

    • 为了减少flush过程对读写的影响，将整个flush过程分为三个阶段：

      • prepare阶段：遍历当前Region中所有的Memstore，将Memstore中当前数据集CellSkipListSet做一个快照snapshot；然后再新建一个CellSkipListSet。后期写入的数据都会写入新的CellSkipListSet中。prepare阶段需要加一把updateLock对写请求阻塞，结束之后会释放该锁。因为此阶段没有任何费时操作，因此持锁时间很短。

      • flush阶段：遍历所有Memstore，将prepare阶段生成的snapshot持久化为临时文件，临时文件会统一放到目录.tmp下。这个过程因为涉及到磁盘IO操作，因此相对比较耗时。

      • commit阶段：遍历所有Memstore，将flush阶段生成的临时文件移到指定的ColumnFamily目录下，针对HFile生成对应的storefile和Reader，把storefile添加到HStore的storefiles列表中，最后再清空prepare阶段生成的snapshot。

    Compact合并机制

    • hbase为了==防止小文件过多==，以保证查询效率，hbase需要在必要的时候将这些小的store file合并成相对较大的store file，这个过程就称之为compaction。

    • 在hbase中主要存在两种类型的compaction合并

      • ==minor compaction 小合并==

      • ==major compaction 大合并==

    minor compaction 小合并

    • 在将Store中多个HFile合并为一个HFile

      在这个过程中会选取一些小的、相邻的StoreFile将他们合并成一个更大的StoreFile，对于超过了==TTL的数据、更新的数据、删除的数据==仅仅只是做了标记。并没有进行物理删除，一次Minor Compaction的结果是更少并且更大的StoreFile。这种合并的触发频率很高。

    • minor compaction触发条件由以下几个参数共同决定：

    

    hbase.hstore.compactionThreshold
    3

​


    hbase.hstore.compaction.max
    10

​


    hbase.hstore.compaction.min.size
    134217728

​


    hbase.hstore.compaction.max.size
    9223372036854775807

    2 major compaction 大合并

    • 合并Store中所有的HFile为一个HFile

      将所有的StoreFile合并成一个StoreFile，这个过程还会清理三类无意义数据：被删除的数据、TTL过期数据、版本号超过设定版本号的数据。合并频率比较低，默认7天执行一次，并且性能消耗非常大，建议生产关闭(设置为0)，在应用空闲时间手动触发。一般可以是手动控制进行合并，防止出现在业务高峰期。

    • major compaction触发时间条件

      

    hbase.hregion.majorcompaction
    604800000

    • 手动触发

      ##使用major_compact命令
major_compact tableName

    HBase表的预分区

    • 当一个table刚被创建的时候，Hbase默认的分配一个region给table。也就是说这个时候，所有的读写请求都会访问到同一个regionServer的同一个region中，这个时候就达不到负载均衡的效果了，集群中的其他regionServer就可能会处于比较空闲的状态。

    • 解决这个问题可以用pre-splitting,在创建table的时候就配置好，生成多个region。

      为何要预分区？

    • 增加数据读写效率

    • 负载均衡，防止数据倾斜

    • 方便集群容灾调度region

    • 优化Map数量

    2、预分区原理

    • 每一个region维护着startRow与endRowKey，如果加入的数据符合某个region维护的rowKey范围，则该数据交给这个region维护。

    3、手动指定预分区

    • 三种方式

    第一种

    create 'person','info1','info2',SPLITS => ['1000','2000','3000','4000'] 

     

    方式二：也可以把分区规则创建于文件中

    cd /kkb/install

    vim split.txt

    aaa
    bbb
    ccc
    ddd 

    create 'student','info',SPLITS_FILE => '/kkb/install/split.txt' 

    方式三： HexStringSplit 算法

    • HexStringSplit会将数据从“00000000”到“FFFFFFFF”之间的数据长度按照n等分之后算出每一段的起始rowkey和结束rowkey，以此作为拆分点。

    • 例如：

    create 'mytable', 'base_info',' extra_info', {NUMREGIONS => 15, SPLITALGO => 'HexStringSplit'} 

    region的冷合并和热合并 

    region合并说明

    • Region的合并不是为了性能, 而是出于便于运维的目的 .

    • 比如删除了大量的数据 ,这个时候每个Region都变得很小 ,存储多个Region就浪费了 ,这个时候可以把Region合并起来，进而可以减少一些Region服务器节点

    一、冷合并 

    这里通过org.apache.hadoop.hbase.util.Merge类来实现，不需要进入hbase shell，直接执行（==需要先关闭hbase集群==）：

    hbase org.apache.hadoop.hbase.util.Merge test test,,1565940912661.62d28d7d20f18debd2e7dac093bc09d8. test,1000,1565940912661.5b6f9e8dad3880bcc825826d12e81436.

    2、通过online_merge热合并Region

    • ==不需要关闭hbase集群==，在线进行合并

    • 与冷合并不同的是，online_merge的传参是Region的hash值，而Region的hash值就是Region名称的最后那段在两个.之间的字符串部分。

    • 需求：需要把test表中的2个region数据进行合并： test,2000,1565940912661.c2212a3956b814a6f0d57a90983a8515. test,3000,1565940912661.553dd4db667814cf2f050561167ca030.

    • 需要进入hbase shell：

      merge_region 'a34185c9f32b1e3211d6d688664266f9','abf69cdc7335880918a60f3c917093e9'

    hbase实战 

    1、hbase实战一（hbase通过MR将一个表中数据，筛选到宁外一张表中）

    • 需求：==读取HBase当中myuser这张表的f1:name、f1:age数据，将数据写入到另外一张myuser2表的f1列族里面去==

    • 第一步：创建myuser2这张hbase表

    注意：列族的名字要与myuser表的列族名字相同

    create 'myuser2','f1'

     HBaseReadMapper类，继承TableMapper，拿到每条数据，并对数据进行过滤并封装

    1. import org.apache.hadoop.hbase.Cell;
    2. import org.apache.hadoop.hbase.CellUtil;
    3. import org.apache.hadoop.hbase.client.Put;
    4. import org.apache.hadoop.hbase.client.Result;
    5. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    6. import org.apache.hadoop.hbase.mapreduce.TableMapper;
    7. import org.apache.hadoop.hbase.util.Bytes;
    8. import org.apache.hadoop.io.Text;
    9.  
    10. import java.io.IOException;
    11.  
    12. /**
    13.  * myuser f1: name&age => myuser2 f1
    14.  */
    15. public class HBaseReadMapper extends TableMapper {
    16.     /**
    17.      *
    18.      * @param key rowkey
    19.      * @param value rowkey此行的数据 Result类型
    20.      * @param context
    21.      * @throws IOException
    22.      * @throws InterruptedException
    23.      */
    24.     @Override
    25.     protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
    26.         //获得roweky的字节数组
    27.         byte[] rowkey_bytes = key.get();
    28.         String rowkeyStr = Bytes.toString(rowkey_bytes);
    29.         Text text = new Text(rowkeyStr);
    30.  
    31.         //输出数据 -> 写数据 -> Put 构建Put对象
    32.         Put put = new Put(rowkey_bytes);
    33.         //获取一行中所有的Cell对象
    34.         Cell[] cells = value.rawCells();
    35.         //将f1 : name& age输出
    36.         for(Cell cell: cells) {
    37.             //当前cell是否是f1
    38.             //列族
    39.             byte[] family_bytes = CellUtil.cloneFamily(cell);
    40.             String familyStr = Bytes.toString(family_bytes);
    41.             if("f1".equals(familyStr)) {
    42.                 //在判断是否是name | age
    43.                 byte[] qualifier_bytes = CellUtil.cloneQualifier(cell);
    44.                 String qualifierStr = Bytes.toString(qualifier_bytes);
    45.                 if("name".equals(qualifierStr)) {
    46.                     put.add(cell);
    47.                 }
    48.                 if("age".equals(qualifierStr)) {
    49.                     put.add(cell);
    50.                 }
    51.             }
    52.         }
    53.  
    54.         //判断是否为空；不为空，才输出
    55.         if(!put.isEmpty()){
    56.             context.write(text, put);
    57.         }
    58.     }
    59. }

     HBaseWriteReducer 类继承TableReducer，合并数据并导出

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    3. import org.apache.hadoop.hbase.mapreduce.TableReducer;
    4. import org.apache.hadoop.io.Text;
    5. import java.io.IOException;
    6.  
    7. /**
    8.  * TableReducer第三个泛型包含rowkey信息
    9.  */
    10. public class HBaseWriteReducer extends TableReducer {
    11.     //将map传输过来的数据，写入到hbase表
    12.     @Override
    13.     protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
    14.         //rowkey
    15.         ImmutableBytesWritable immutableBytesWritable = new ImmutableBytesWritable();
    16.         immutableBytesWritable.set(key.toString().getBytes());
    17.  
    18.         //遍历put对象，并输出
    19.         for(Put put: values) {
    20.             context.write(immutableBytesWritable, put);
    21.         }
    22.     }
    23. }

    main方法入口 

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.conf.Configured;
    3. import org.apache.hadoop.hbase.HBaseConfiguration;
    4. import org.apache.hadoop.hbase.TableName;
    5. import org.apache.hadoop.hbase.client.Put;
    6. import org.apache.hadoop.hbase.client.Scan;
    7. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
    8. import org.apache.hadoop.io.Text;
    9. import org.apache.hadoop.mapreduce.Job;
    10. import org.apache.hadoop.util.Tool;
    11. import org.apache.hadoop.util.ToolRunner;
    12.  
    13. public class HBaseMR extends Configured implements Tool {
    14.     public static void main(String[] args) throws Exception {
    15.         Configuration configuration = HBaseConfiguration.create();
    16.         //设定绑定的zk集群
    17.         configuration.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181");
    18.  
    19.         int run = ToolRunner.run(configuration, new HBaseMR(), args);
    20.         System.exit(run);
    21.     }
    22.  
    23.     @Override
    24.     public int run(String[] args) throws Exception {
    25.         Job job = Job.getInstance(super.getConf());
    26.         job.setJarByClass(HBaseMR.class);
    27.  
    28.         //mapper
    29.         TableMapReduceUtil.initTableMapperJob(TableName.valueOf("myuser"), new Scan(),HBaseReadMapper.class, Text.class, Put.class, job);
    30.         //reducer
    31.         TableMapReduceUtil.initTableReducerJob("myuser2", HBaseWriteReducer.class, job);
    32.  
    33.         boolean b = job.waitForCompletion(true);
    34.         return b? 0: 1;
    35.     }
    36. }

     实战二，读取HDFS上的数据，并将数据写到Hbase中

    需求读取hdfs上面的数据，写入到hbase表里面去

    node03执行以下命令准备数据文件，并将数据文件上传到HDFS上面去

    在/kkb/install目录，创建user.txt文件

    cd /kkb/install
    vim user.txt

    数据内容 

    0007    zhangsan    18
    0008    lisi    25
    0009    wangwu    20

    创建HDFS上文件夹和将本地文件导入到文件夹中 

    hdfs dfs -mkdir -p /hbase/input
    hdfs dfs -put /kkb/install/user.txt /hbase/input/

    删除hdfs上的文件，（与实战二无关） 

    hdfs dfs -rm /hbase/input/user.txt

    1. package com.kkb.hbase.hdfs2hbase;
    2.  
    3. import org.apache.hadoop.conf.Configuration;
    4. import org.apache.hadoop.fs.Path;
    5. import org.apache.hadoop.hbase.HBaseConfiguration;
    6. import org.apache.hadoop.hbase.client.Put;
    7. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    8. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
    9. import org.apache.hadoop.hbase.mapreduce.TableReducer;
    10. import org.apache.hadoop.hbase.util.Bytes;
    11. import org.apache.hadoop.io.LongWritable;
    12. import org.apache.hadoop.io.NullWritable;
    13. import org.apache.hadoop.io.Text;
    14. import org.apache.hadoop.mapreduce.Job;
    15. import org.apache.hadoop.mapreduce.Mapper;
    16. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    17.  
    18. import java.io.IOException;
    19.  
    20. /**
    21.  * 将HDFS上文件/hbase/input/user.txt数据，导入到HBase的myuser2表
    22.  */
    23. public class HDFS2HBase {
    24.     public static class HdfsMapper extends Mapper {
    25.  
    26.         //数据原样输出
    27.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    28.             context.write(value, NullWritable.get());
    29.         }
    30.     }
    31.  
    32.     public static class HBaseReducer extends TableReducer {
    33.  
    34.         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
    35.             /**
    36.              * key -> 一行数据
    37.              * 样例数据：
    38.              * 0007	zhangsan	18
    39.              * 0008	lisi	25
    40.              * 0009	wangwu	20
    41.              */
    42.             String[] split = key.toString().split("\t");
    43.  
    44.             Put put = new Put(Bytes.toBytes(split[0]));
    45.             put.addColumn("f1".getBytes(), "name".getBytes(), split[1].getBytes());
    46.             put.addColumn("f1".getBytes(), "age".getBytes(), split[2].getBytes());
    47.  
    48.             context.write(new ImmutableBytesWritable(Bytes.toBytes(split[0])), put);
    49.         }
    50.     }
    51.  
    52.     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    53.         Configuration conf = HBaseConfiguration.create();
    54.         //设定zk集群
    55.         conf.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181");
    56.         Job job = Job.getInstance(conf);
    57.  
    58.         job.setJarByClass(HDFS2HBase.class);
    59.  
    60.         //可省略
    61.         //job.setInputFormatClass(TextInputFormat.class);
    62.         //输入文件路径
    63.         FileInputFormat.addInputPath(job, new Path("hdfs://node01:8020/hbase/input"));
    64.  
    65.         job.setMapperClass(HdfsMapper.class);
    66.         //map端的输出的key value 类型
    67.         job.setMapOutputKeyClass(Text.class);
    68.         job.setMapOutputValueClass(NullWritable.class);
    69.  
    70.         //指定输出到hbase的表名
    71.         TableMapReduceUtil.initTableReducerJob("myuser2", HBaseReducer.class, job);
    72.  
    73.         //设置reduce个数
    74.         job.setNumReduceTasks(1);
    75.  
    76.         System.exit(job.waitForCompletion(true) ? 0 : 1);
    77.     }
    78. }

    3、实战三（将HDFS上的文件转化为Hfile，然后再hbase刷新数据）

    • 需求

      • ==通过bulkload的方式批量加载数据到HBase表中==

      • ==将我们hdfs上面的这个路径/hbase/input/user.txt的数据文件，转换成HFile格式，然后load到myuser2这张表里面去==

    • 知识点描述

      • 加载数据到HBase当中去的方式多种多样，我们可以使用HBase的javaAPI或者使用sqoop将我们的数据写入或者导入到HBase当中去，但是这些方式不是慢就是在导入的过程的占用Region资源导致效率低下

      • 我们也可以通过MR的程序，将我们的数据直接转换成HBase的最终存储格式HFile，然后直接load数据到HBase当中去即可

    好处

    • 导入过程不占用Region资源

    • 能快速导入海量的数据

    • 节省内存

    继承 Mapper，输出key为空的put对象，将hdfs文件转化为hfile格式文件

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    3. import org.apache.hadoop.io.LongWritable;
    4. import org.apache.hadoop.io.Text;
    5. import org.apache.hadoop.mapreduce.Mapper;
    6.  
    7. import java.io.IOException;
    8.  
    9. //四个泛型中后两个，分别对应rowkey及put
    10. public class BulkLoadMapper extends Mapper {
    11.     @Override
    12.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    13.         String[] split = value.toString().split("\t");
    14.         //封装输出的rowkey类型
    15.         ImmutableBytesWritable immutableBytesWritable = new ImmutableBytesWritable(split[0].getBytes());
    16.  
    17.         //构建put对象
    18.         Put put = new Put(split[0].getBytes());
    19.         put.addColumn("f1".getBytes(), "name".getBytes(), split[1].getBytes());
    20.         put.addColumn("f1".getBytes(), "age".getBytes(), split[2].getBytes());
    21.  
    22.         context.write(immutableBytesWritable, put);
    23.     }
    24. }

    main主程序

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.conf.Configured;
    3. import org.apache.hadoop.fs.Path;
    4. import org.apache.hadoop.hbase.HBaseConfiguration;
    5. import org.apache.hadoop.hbase.TableName;
    6. import org.apache.hadoop.hbase.client.Connection;
    7. import org.apache.hadoop.hbase.client.ConnectionFactory;
    8. import org.apache.hadoop.hbase.client.Put;
    9. import org.apache.hadoop.hbase.client.Table;
    10. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    11. import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
    12. import org.apache.hadoop.mapreduce.Job;
    13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    14. import org.apache.hadoop.util.Tool;
    15. import org.apache.hadoop.util.ToolRunner;
    16.  
    17. public class HBaseBulkLoad extends Configured implements Tool {
    18.     public static void main(String[] args) throws Exception {
    19.         Configuration configuration = HBaseConfiguration.create();
    20.         //设定zk集群
    21.         configuration.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181");
    22.  
    23.         int run = ToolRunner.run(configuration, new HBaseBulkLoad(), args);
    24.         System.exit(run);
    25.     }
    26.     @Override
    27.     public int run(String[] args) throws Exception {
    28.         Configuration conf = super.getConf();
    29.         Job job = Job.getInstance(conf);
    30.         job.setJarByClass(HBaseBulkLoad.class);
    31.  
    32.         FileInputFormat.addInputPath(job, new Path("hdfs://node01:8020/hbase/input"));
    33.         job.setMapperClass(BulkLoadMapper.class);
    34.         job.setMapOutputKeyClass(ImmutableBytesWritable.class);
    35.         job.setMapOutputValueClass(Put.class);
    36.  
    37.         Connection connection = ConnectionFactory.createConnection(conf);
    38.         Table table = connection.getTable(TableName.valueOf("myuser2"));
    39.  
    40.         //使MR可以向myuser2表中，增量增加数据
    41.         HFileOutputFormat2.configureIncrementalLoad(job, table, connection.getRegionLocator(TableName.valueOf("myuser2")));
    42.         //数据写回到HDFS，写成HFile -> 所以指定输出格式为HFileOutputFormat2
    43.         job.setOutputFormatClass(HFileOutputFormat2.class);
    44.         HFileOutputFormat2.setOutputPath(job, new Path("hdfs://node01:8020/hbase/out_hfile"));
    45.  
    46.         //开始执行
    47.         boolean b = job.waitForCompletion(true);
    48.  
    49.         return b? 0: 1;
    50.     }
    51. }

     

    将处理好的HDFS文件转化为 Hbase中的Hfile文件

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.fs.Path;
    3. import org.apache.hadoop.hbase.HBaseConfiguration;
    4. import org.apache.hadoop.hbase.TableName;
    5. import org.apache.hadoop.hbase.client.Admin;
    6. import org.apache.hadoop.hbase.client.Connection;
    7. import org.apache.hadoop.hbase.client.ConnectionFactory;
    8. import org.apache.hadoop.hbase.client.Table;
    9. import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
    10.  
    11. public class LoadData {
    12.     public static void main(String[] args) throws Exception {
    13.         Configuration configuration = HBaseConfiguration.create();
    14.         configuration.set("hbase.zookeeper.quorum", "node01,node02,node03");
    15.         //获取数据库连接
    16.         Connection connection = ConnectionFactory.createConnection(configuration);
    17.         //获取表的管理器对象
    18.         Admin admin = connection.getAdmin();
    19.         //获取table对象
    20.         TableName tableName = TableName.valueOf("myuser2");
    21.         Table table = connection.getTable(tableName);
    22.         //构建LoadIncrementalHFiles加载HFile文件
    23.         LoadIncrementalHFiles load = new LoadIncrementalHFiles(configuration);
    24.         load.doBulkLoad(new Path("hdfs://node01:8020/hbase/out_hfile"), admin, table, connection.getRegionLocator(tableName));
    25.     }
    26. }

    加载之前数据

    加载后的数据

    HBase集成Hive

    Hive提供了与HBase的集成，使得能够在HBase表上使用hive sql 语句进行查询、插入操作以及进行Join和Union等复杂查询，同时也可以将hive表中的数据映射到Hbase中

    HBase与Hive的对比

    .1 Hive

    • 数据仓库

      Hive的本质其实就相当于将HDFS中已经存储的文件在Mysql中做了一个双射关系，以方便使用HQL去管理查询。

    • 用于数据分析、清洗

      Hive适用于离线的数据分析和清洗，延迟较高

    • 基于HDFS、MapReduce

      Hive存储的数据依旧在DataNode上，编写的HQL语句终将是转换为MapReduce代码执行。（不要钻不需要执行MapReduce代码的情况的牛角尖）

    2、HBase

    • 数据库

      是一种面向列存储的非关系型数据库。

    • 用于存储结构化和非结构话的数据

      适用于单表非关系型数据的存储，不适合做关联查询，类似JOIN等操作。

    • 基于HDFS

      数据持久化存储的体现形式是Hfile，存放于DataNode中，被ResionServer以region的形式进行管理。

    • 延迟较低，接入在线业务使用

      面对大量的企业数据，HBase可以直线单表大量数据的存储，同时提供了高效的数据访问速度。

    3、总结：Hive与HBase

    • Hive和Hbase是两种基于Hadoop的不同技术，Hive是一种类SQL的引擎，并且运行MapReduce任务，Hbase是一种在Hadoop之上的NoSQL 的Key/vale数据库。这两种工具是可以同时使用的。就像用Google来搜索，用FaceBook进行社交一样，Hive可以用来进行统计查询，HBase可以用来进行实时查询，数据也可以从Hive写到HBase，或者从HBase写回Hive。

    hbase导入HIve环境整合 

    拷贝jar包

    • 将我们HBase的五个jar包拷贝到hive的lib目录下

    • hbase的jar包都在/kkb/install/hbase-1.2.0-cdh5.14.2/lib

    • 我们需要拷贝五个jar包名字如下

     hbase-client-1.2.0-cdh5.14.2.jar                  
    hbase-hadoop2-compat-1.2.0-cdh5.14.2.jar 
    hbase-hadoop-compat-1.2.0-cdh5.14.2.jar  
    hbase-it-1.2.0-cdh5.14.2.jar    
    hbase-server-1.2.0-cdh5.14.2.jar

    ln -s /kkb/install/hbase-1.2.0-cdh5.14.2/lib/hbase-client-1.2.0-cdh5.14.2.jar              /kkb/install/hive-1.1.0-cdh5.14.2/lib/hbase-client-1.2.0-cdh5.14.2.jar   

    ln -s /kkb/install/hbase-1.2.0-cdh5.14.2/lib/hbase-hadoop2-compat-1.2.0-cdh5.14.2.jar      /kkb/install/hive-1.1.0-cdh5.14.2/lib/hbase-hadoop2-compat-1.2.0-cdh5.14.2.jar    

            
    ln -s /kkb/install/hbase-1.2.0-cdh5.14.2/lib/hbase-hadoop-compat-1.2.0-cdh5.14.2.jar       /kkb/install/hive-1.1.0-cdh5.14.2/lib/hbase-hadoop-compat-1.2.0-cdh5.14.2.jar    

           
    ln -s /kkb/install/hbase-1.2.0-cdh5.14.2/lib/hbase-it-1.2.0-cdh5.14.2.jar     /kkb/install/hive-1.1.0-cdh5.14.2/lib/hbase-it-1.2.0-cdh5.14.2.jar    

    ln -s /kkb/install/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar          /kkb/install/hive-1.1.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar  

    修改hive的配置文件

    • 编辑node03服务器上面的hive的配置文件hive-site.xml

     cd /kkb/install/hive-1.1.0-cdh5.14.2/conf


    vim hive-site.xml

    添加如下配置 

        
            hive.zookeeper.quorum
            node01,node02,node03
        
        
            hbase.zookeeper.quorum
            node01,node02,node03

    修改hive-env.sh文件，添加hbase环境 

    cd /kkb/install/hive-1.1.0-cdh5.14.2/conf
    vim hive-env.sh 

    export HBASE_HOME=/kkb/install/hbase-1.2.0-cdh5.14.2 

    需求一：将hive表当中分析的结果保存到hbase表当中去

    1、hive中创建一个外部表

    create database course;


    use course;

    create external table if not exists course.score(id int, cname string, score int) 
    row format delimited fields terminated by '\t' stored as textfile ;

    2、创建数据导入表中

    cd /kkb/install/hivedatas
    vim hive-hbase.txt

    1    zhangsan    80
    2    lisi    60
    3    wangwu    30
    4    zhaoliu    70 

    本地导入数据

    hive (course)> load data local inpath '/kkb/install/hivedatas/hive-hbase.txt' into table score;


    hive (course)> select * from score; 

    创建hive管理表与HBase进行映射

    • 我们可以创建一个hive的管理表与hbase当中的表进行映射，hive管理表当中的数据，都会存储到hbase上面去

    • hive当中创建内部表

    create table course.hbase_score(id int,cname string,score int) 
    stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'  
    with serdeproperties("hbase.columns.mapping" = ":key,cf:name,cf:score") tblproperties("hbase.table.name" = "hbase_score"); 

    stored by：org.apache.hadoop.hive.hbase.HBaseStorageHandler（存储方式） 

    with serdeproperties("hbase.columns.mapping" = ":key,cf:name,cf:score") （hive表中字段与hbase中的映射关系，id映射:key,，cname映射name，score映射score）

     tblproperties("hbase.table.name" = "hbase_score"); 映射hbase中的表，对应的表名为hbase_score

    通过insert overwrite select 插入数据

    insert overwrite table course.hbase_score select id,cname,score from course.score;

    如果插入数据权限不够，则赋权 

    hdfs dfs -chown -R root:root /tmp 

    查看hive中的数据

     查看hbase中的数据

    需求二：创建hive外部表，映射HBase当中已有的表模型

    1、hbase中创建表

    # 创建一张表
    create 'hbase_hive_score',{ NAME =>'cf'}
    # 通过put插入数据到hbase表
    put 'hbase_hive_score','1','cf:name','zhangsan'
    put 'hbase_hive_score','1','cf:score', '95'
    put 'hbase_hive_score','2','cf:name','lisi'
    put 'hbase_hive_score','2','cf:score', '96'
    put 'hbase_hive_score','3','cf:name','wangwu'
    put 'hbase_hive_score','3','cf:score', '97'

    建立hive的外部表，映射HBase当中的表以及字段

    • 在hive当中建立外部表

    • 进入hive客户端，然后执行以下命令进行创建hive外部表，就可以实现映射HBase当中的表数据

     CREATE external TABLE course.hbase2hive(id int, name string, score int) 
    STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 
    WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf:name,cf:score") TBLPROPERTIES("hbase.table.name" ="hbase_hive_score");

    因为定义了映射关系，所以直接查询

    select * from course.hbase2hive; 

     

    Phoenix安装使用

    Phoenix安装

    安装包链接

    • 从对应的地址下载：Index of /dist/phoenix

    • 这里我们使用的是

      • apache-phoenix-4.14.0-cdh5.14.2-bin.tar.gz

    1、上传文件并解压

    cd /kkb/soft/
    tar -zxf apache-phoenix-4.14.0-cdh5.14.2-bin.tar.gz  -C /kkb/install/

     2、修改配置

    • 1、将phoenix目录下的==phoenix-4.8.2-HBase-1.2-server.jar==、

      ==phoenix-core-4.8.2-HBase-1.2.jar==拷贝到==各个 hbase的lib目录==下。

      ==node02执行==以下命令，将两个jar包拷贝到hbase的lib目录下 

      ​
cd /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin


scp phoenix-4.14.0-cdh5.14.2-server.jar phoenix-core-4.14.0-cdh5.14.2.jar node01:/kkb/install/hbase-1.2.0-cdh5.14.2/lib/ 
​
scp phoenix-4.14.0-cdh5.14.2-server.jar phoenix-core-4.14.0-cdh5.14.2.jar node02:/kkb/install/hbase-1.2.0-cdh5.14.2/lib/ 
​
scp phoenix-4.14.0-cdh5.14.2-server.jar phoenix-core-4.14.0-cdh5.14.2.jar node03:/kkb/install/hbase-1.2.0-cdh5.14.2/lib/

    2、 将hbase的配置文件==hbase-site.xml==、 hadoop下的配置文件==core-site.xml== 、==hdfs-site.xml==放到phoenix/bin/下，替换phoenix原来的配置文件。

    ==node02执行==以下命令，进行拷贝配置文件

    cp /kkb/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop/core-site.xml  /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/bin/

    cp /kkb/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop/hdfs-site.xml  /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/bin/

    cp /kkb/install/hbase-1.2.0-cdh5.14.2/conf/hbase-site.xml  /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/bin/

     

    3、重启hbase集群，使Phoenix的jar包生效。

    • 记得要先启动hadoop集群、zookeeper集群

    • ==node01执行==以下命令来重启hbase的集群

     4、检查是否安装成功

    在bin目录下，执行bin/sqlline.py node01:2181

     cd /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/
    bin/sqlline.py node01:2181

     这样就是链接好了

    help 

    查看帮助文档

     

     

    Phoenix使用

    批处理方式，批量处理SQL

    这演示，创建SQL，SQL数据，SQL查询，的批量处理

    1、创建一个创建SQL表的sql

    node02执行以下命令创建user_phoenix.sql文件

    mkdir -p /kkb/install/phoenixsql


    cd /kkb/install/phoenixsql/


     sudo vim user_phoenix.sql

    create table if not exists user_phoenix (state varchar(10) NOT NULL,  city varchar(20) NOT NULL, population BIGINT  CONSTRAINT my_pk PRIMARY KEY (state, city));

    2、创建数据

    node02执行以下命令，创建user_phoenix.csv数据文件

    cd /kkb/install/phoenixsql/


    vim user_phoenix.csv

    NY,New York,8143197
    CA,Los Angeles,3844829
    IL,Chicago,2842518
    TX,Houston,2016582
    PA,Philadelphia,1463281
    AZ,Phoenix,1461575
    TX,San Antonio,1256509
    CA,San Diego,1255540
    TX,Dallas,1213825
    CA,San Jose,912332

    3、创建查询的user_phoenix_query.sql

    创建user_phoenix_query.sql文件

    cd /kkb/install/phoenixsql


    vim user_phoenix_query.sql

    select state as "userState",count(city) as "City Count",sum(population) as "Population Sum" FROM user_phoenix GROUP BY state; 

     4、运行

    cd /kkb/install/phoenixsql

    /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/bin/psql.py  node01:2181 user_phoenix.sql user_phoenix.csv  user_phoenix_query.sql

     Phoenix中查询的表数据

     

    Hbase中查询的表数据

     

     命令方式实现phoenix和hbase的映射

    1、启动phoenix

    cd /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/
    bin/sqlline.py node01:2181

    退出 

     !quit

    查看所有的表 

    !table 

    2、建立employee的映射表

    进入hbase客户端，创建一个普通表employee，并且有两个列族 company 和family node01执行以下以下命令进入hbase 的shell客户端

     hbase(main):001:0> create 'employee','company','family'

    插入数据 

    put 'employee','row1','company:name','ted'
    put 'employee','row1','company:position','worker'
    put 'employee','row1','family:tel','13600912345'
    put 'employee','row1','family:age','18'
    put 'employee','row2','company:name','michael'
    put 'employee','row2','company:position','manager'
    put 'employee','row2','family:tel','1894225698'
    put 'employee','row2','family:age','20' 

    3、建立hbase到phoenix的映射表

    node02进入到phoenix的客户端，然后创建映射表

    cd /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin
    bin/sqlline.py node01:2181

    CREATE TABLE IF NOT EXISTS "employee" ("no" VARCHAR(10) NOT NULL PRIMARY KEY, "company"."name" VARCHAR(30),"f1"."age" VARCHAR(20), "family"."tel" VARCHAR(20), "family"."age" VARCHAR(20)) column_encoded_bytes=0;

     

    说明

    在建立映射表之前要说明的是，Phoenix是==大小写敏感==的，并且所有命令都是大写

    如果你建的表名没有用双引号括起来，那么无论你输入的是大写还是小写，建立出来的表名都是大写的

    如果你需要建立出同时包含大写和小写的表名和字段名，请把表名或者字段名用==双引号括起来==。

     select * from "employee";

     

     select * from "employee" where "tel" = '13600912345';

     

    JDBC实现javaAPI对Phoenix数据查询 

     

    1. <dependencies>
    2.     <dependency>
    3.         <groupId>org.apache.phoenixgroupId>
    4.         <artifactId>phoenix-coreartifactId>
    5.         <version>4.14.0-cdh5.14.2version>
    6.     dependency>
    7.     <dependency>
    8.         <groupId>junitgroupId>
    9.         <artifactId>junitartifactId>
    10.         <version>4.12version>
    11.     dependency>
    12.     <dependency>
    13.         <groupId>org.testnggroupId>
    14.         <artifactId>testngartifactId>
    15.         <version>6.14.3version>
    16.     dependency>
    17.     dependencies>
    18.     <build>
    19.         <plugins>
    20.             
    21.             <plugin>
    22.                 <groupId>org.apache.maven.pluginsgroupId>
    23.                 <artifactId>maven-compiler-pluginartifactId>
    24.                 <version>3.0version>
    25.                 <configuration>
    26.                     <source>1.8source>
    27.                     <target>1.8target>
    28.                     <encoding>UTF-8encoding>
    29.                 configuration>
    30.             plugin>
    31.         plugins>
    32.     build>

    java代码

    1. import org.testng.annotations.BeforeTest;
    2. import org.testng.annotations.Test;
    3. import java.sql.*;
    4.  
    5. public class PhoenixSearch {
    6.     private Connection connection;
    7.     private Statement statement;
    8.     private ResultSet rs;
    9.     
    10.     @BeforeTest
    11.     public void init() throws SQLException {
    12.         //定义phoenix的连接url地址
    13.         String url="jdbc:phoenix:node01:2181";
    14.         connection = DriverManager.getConnection(url);
    15.         //构建Statement对象
    16.         statement = connection.createStatement();
    17.     }
    18.     @Test
    19.     public void queryTable() throws SQLException {
    20.         //定义查询的sql语句，注意大小写
    21.         String sql="select * from US_POPULATION";
    22.         //执行sql语句
    23.         try {
    24.             rs=statement.executeQuery(sql);
    25.             while(rs.next()){
    26.                 System.out.println("state:"+rs.getString("state"));
    27.                 System.out.println("city:"+rs.getString("city"));
    28.                 System.out.println("population:"+rs.getInt("population"));
    29.                 System.out.println("-------------------------");
    30.             }
    31.         } catch (SQLException e) {
    32.             e.printStackTrace();
    33.         }finally {
    34.             if(connection!=null){
    35.                 connection.close();
    36.             }
    37.         }
    38.     }
    39. }

     打印的值为：

     

    Phoenix构建二级索引

    • 对于HBase而言，如果想精确地定位到某行记录，唯一的办法是通过rowkey来查询。如果不通过rowkey来查找数据，就必须逐行地比较每一列的值，即全表扫瞄。

    • 对于较大的表，全表扫描的代价是不可接受的。但是，很多情况下，需要从多个角度查询数据。

      • 例如，在定位某个人的时候，可以通过姓名、身份证号、学籍号等不同的角度来查询

      • 要想把这么多角度的数据都放到rowkey中几乎不可能（业务的灵活性不允许，对rowkey长度的要求也不允许）。

      • 所以需要secondary index（二级索引）来完成这件事。secondary index的原理很简单，但是如果自己维护的话则会麻烦一些。

      • 现在，Phoenix已经提供了对HBase secondary index的支持。

    启动二级索引功能 

    修改配置文件

    • 如果要启用phoenix的二级索引功能，需要修改配置文件hbase-site.xml

    • 注意：

     cd /kkb/install/hbase-1.2.0-cdh5.14.2/conf/

    修改配置文件 

    vim   hbase-site.xml 

    添加下面配置 

    1. <property>
    2. 	<name>hbase.regionserver.wal.codecname> 
    3.     <value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodecvalue>
    4.  property>
    5.  <property>
    6.     <name>hbase.region.server.rpc.scheduler.factory.classname>
    7.     <value>org.apache.hadoop.hbase.ipc.PhoenixRpcSchedulerFactoryvalue>
    8. property>
    9. <property>
    10. 	<name>hbase.rpc.controllerfactory.classname>
    11.     <value>org.apache.hadoop.hbase.ipc.controller.ServerRpcControllerFactoryvalue>
    12. property>

    拷贝配置到node02和node03当前配置项

    scp hbase-site.xml  node02:$PWD 

    scp hbase-site.xml  node03:$PWD

    拷贝完后，重启hbase集群

    数据导入

    • 首先，在phoenix中创建一个user table

    • node02执行以下命令，进入phoenix客户端，并创建表

    create  table user (
    "session_id" varchar(100) not null primary key, 
    "f"."cookie_id" varchar(100), 
    "f"."visit_time" varchar(100), 
    "f"."user_id" varchar(100), 
    "f"."age" varchar(100), 
    "f"."sex" varchar(100), 
    "f"."visit_url" varchar(100), 
    "f"."visit_os" varchar(100), 
    "f"."browser_name" varchar(100),
    "f"."visit_ip" varchar(100), 
    "f"."province" varchar(100),
    "f"."city" varchar(100),
    "f"."page_id" varchar(100), 
    "f"."goods_id" varchar(100),
    "f"."shop_id" varchar(100)) column_encoded_bytes=0;  

     Phoenix中查看表

    在hbase中查看可知，hbase中也产生了一个USER表

    准备数据并导入

    node02执行以下命令，导入50W的测试数据

    cd /kkb/install/apache-phoenix-4.14.0-cdh5.14.2-bin/
    bin/psql.py -t USER node01:2181 /kkb/install/phoenixsql/user50w.csv 

    测试导入数据

    查询user前三条数据

    SELECT * FROM user LIMIT 3;

     

    Phoenix 的全局索引（Global Indexing） 和本地索引（Local Indexing）

    1、Global Indexing （全局索引）

    • Global indexing，全局索引，适用于==读多写少==的业务场景。

    • 使用Global indexing在写数据的时候开销很大，因为所有对数据表的更新操作（DELETE, UPSERT VALUES and UPSERT SELECT），都会引起索引表的更新，而索引表是分布在不同的数据节点上的，跨节点的数据传输带来了较大的性能消耗。

    • 在读数据的时候Phoenix会选择索引表来降低查询消耗的时间。

      • 在默认情况下如果想查询的字段不是索引字段的话索引表不会被使用，也就是说不会带来查询速度的提升。

     不加索引的情况下，执行下面SQL，运行时间98秒（本来电脑就不好）

    select "cookie_id" from user where "cookie_id" = '99738fd1-2084-44e9';

    通过Explian来分析SQL语句

    explain select * from user where "cookie_id" = '99738fd1-2084-44e9';

     FULL SCAN OVER USER（全表扫描的方式）

    给表USER创建基于Global Indexing的二级索引

    jdbc:phoenix:node01:2181> create index USER_COOKIE_ID_INDEX on USER ("f"."cookie_id"); 

     select * from USER_COOKIE_ID_INDEX limit 5;

     再执行查询操作

    select "cookie_id" from user where "cookie_id" = '99738fd1-2084-44e9';

    时间降低为0.5秒 

    分析语句，由此可看是走索引的（ SCAN OVER USER_COOKIE_ID_INDEX ['99738fd1-2084-44e9']）

    explain select "cookie_id" from user where "cookie_id"='99738fd1-2084-44e9';

     但是注意

    select "cookie_id","age" from user where "cookie_id"='99738fd1-2084-44e9';

    这个因为其他字段，没有建立索引，所以如果走全局索引，如果查询没有索引的字段，该SQL语句还是全表扫描，可以看到查询时间6秒

     采用explain分析sql （ ROUND ROBIN FULL SCAN OVER USER ）

     explain select "cookie_id","age" from user where "cookie_id"='99738fd1-2084-44e9'; 

     同理查询其他字段也不会走索引

    select "sex" from user where "cookie_id"='99738fd1-2084-44e9';

     

     

     

    Local Indexing(本地索引)

    • Local indexing，本地索引，适用于==写操作频繁==以及空间受限制的场景。

    • 与Global indexing一样，Phoenix会自动判定在进行查询的时候是否使用索引。

    • 使用Local indexing时，索引数据和数据表的数据存放在相同的服务器中，这样避免了在写操作的时候往不同服务器的索引表中写索引带来的额外开销。

    • 使用Local indexing的时候即使查询的字段不是索引字段索引表也会被使用，这会带来查询速度的提升，这点跟Global indexing不同。对于Local Indexing，一个数据表的所有索引数据都存储在一个单一的独立的可共享的表中。

    测试查询

     select * from user where "user_id"='371e963d-c-487065';

     查看查询情况 

    分析查询语句（WAY ROUND ROBIN FULL SCAN OVER USER）

     explain select * from user where "user_id"='371e963d-c-487065';

     创建本地索引

    create local index USER_USER_ID_INDEX on USER ("f"."user_id");

     

     查询前五条数据

     select * from USER_USER_ID_INDEX limit 5;

     

    通过user_id查询整行数据 ，执行时间5.7秒

    select * from user where "user_id"='371e963d-c-487065'; 

     

     采用explain分析SQL （RANGE SCAN OVER USER [1,'371e963d-c-487065']）可以看到是走的索引，只是我电脑较卡原因

    explain select * from user where "user_id"='371e963d-c-487065';

    非索引字段的查询也很快 ，0.46秒

     select 'user_id','age' from user where "user_id"='371e963d-c-487065';

    分析语句可以查看到是走的索引字段 

    explain  select 'user_id','age' from user where "user_id"='371e963d-c-487065'; 

    如何确保query查询使用Index

    • 要想让一个查询使用index，有三种方式实现。

     创建 covered index

    • 如果在某次查询中，查询项或者查询条件中包含除被索引列之外的列（主键MY_PK除外）。

    • 默认情况下，该查询会触发full table scan（全表扫描），但是使用covered index则可以避免全表扫描。创建包含某个字段的覆盖索引,创建方式如下：

    1、针对表创建多个全局索引 

     create index USER_COOKIE_ID_AGE_INDEX on USER ("f"."cookie_id") include("f"."age");

    查询该表

    select * from USER_COOKIE_ID_AGE_INDEX limit 5;  

    查询age 

    select "age" from user where "cookie_id"='99738fd1-2084-44e9'; 

     查询age 和cookie_id

    select "age","cookie_id" from user where "cookie_id"='99738fd1-2084-44e9';

    通过explain分析两个SQL，发现两个SQL均走的索引

     但是全局索引，查询非索引字段，则不是走索引

    select "age","sex" from user where "cookie_id"='99738fd1-2084-44e9';

    耗时8秒 

     

    explain select "age","sex" from user where "cookie_id"='99738fd1-2084-44e9';

     

    2、在查询中提示其使用index

    • 在select和column_name之间加上/*+ Index(<表名> )*/，通过这种方式强制使用索引。

    select /*+ index(user,USER_COOKIE_ID_AGE_INDEX) */ "age" from user where "cookie_id"='99738fd1-2084-44e9';

    如果age是索引字段，那么就会直接从索引表中查询 如果age不是索引字段，那么将会进行全表扫描，所以当用户明确知道表中数据较少且符合检索条件时才适用，此时的性能才是最佳的。

     3、使用本地索引

    使用本地索引，则通过索引字段，查询其他非索引字段的值时候，也会走索引

    索引重建

    • Phoenix的索引重建是把索引表清空后重新装配数据。

    alter index USER_COOKIE_ID_INDEX on user rebuild;

    删除索引

    • 删除某个表的某张索引： 语法 drop index 索引名称 on 表名 例如：

    drop index USER_COOKIE_ID_INDEX on user;

    如果表中的一个索引列被删除，则索引也将被自动删除

    如果删除的是覆盖索引上的列，则此列将从覆盖索引中被自动删除。

     

    索引性能调优

    • 一般来说，索引已经很快了，不需要特别的优化。

    • 这里也提供了一些方法，让你在面对特定的环境和负载的时候可以进行一些调优。下面的这些需要在hbase-site.xml文件中设置，针对所有的服务器。

     

    1. index.builder.threads.max 
    创建索引时，使用的最大线程数。 
    默认值: 10。

    2. index.builder.threads.keepalivetime 
    创建索引的创建线程池中线程的存活时间，单位：秒。 
    默认值: 60

    3. index.writer.threads.max 
    写索引表数据的写线程池的最大线程数。 
    更新索引表可以用的最大线程数，也就是同时可以更新多少张索引表，数量最好和索引表的数量一致。 
    默认值: 10

    4. index.writer.threads.keepalivetime 
    索引写线程池中，线程的存活时间，单位：秒。
    默认值：60
     
    5. hbase.htable.threads.max 
    每一张索引表可用于写的线程数。 
    默认值: 2,147,483,647

    6. hbase.htable.threads.keepalivetime 
    索引表线程池中线程的存活时间，单位：秒。 
    默认值: 60

    7. index.tablefactory.cache.size 
    允许缓存的索引表的数量。 
    增加此值，可以在写索引表时不用每次都去重复的创建htable，这个值越大，内存消耗越多。 
    默认值: 10

    8. org.apache.phoenix.regionserver.index.handler.count 
    处理全局索引写请求时，可以使用的线程数。 
    默认值: 30

  • 相关阅读:
    聚观早报｜蔚来汽车首颗自研芯片；中式汉堡正打破“麦门永存”
    deque（双端数组）——STL
    IDEA绿色版本重装系统之后git远程仓出现的问题
    Docker入门开发
    NeuralProphet之八：NeuralProphet部署
    bootstrap.properties中配置Nacos
    如何优雅地画一张图
    树状数组——一个简单的整数问题2
    Redis 之高可用与持久化
    AJAX——AJAX的异步与同步、AJAX代码封装
  • 原文地址：https://blog.csdn.net/weixin_43288858/article/details/125785238