HBase(十二)Region 分裂策略

region中存储的是一张表的数据,当region中的数据条数过多的时候,会直接影响查询效率.

当region过大的时候,hbase会将region拆分为两个region , 这也是Hbase的一个优点。

HBase的region split策略一共有以下6种:

1、ConstantSizeRegionSplitPolicy

0.94版本前,HBase region的默认切分策略

当region中最大的store大小超过某个阈值(hbase.hregion.max.filesize=10G)之后就会触发切分,一个region等分为2个region。
但是在生产线上这种切分策略却有相当大的弊端:

  • 切分策略对于大表和小表没有明显的区分。
  • 阈值(hbase.hregion.max.filesize)设置较大对大表比较友好,但是小表就有可能不会触发分裂,极端情况下可能就1个,形成热点,这对业务来说并不是什么好事。
  • 如果设置较小则对小表友好,但一个大表就会在整个集群产生大量的region,这对于集群的管理、资源使用、failover来说都不是一件好事。
2、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上
3、SteppingSplitPolicy

2.0+版本默认切分策略

相比 IncreasingToUpperBoundRegionSplitPolicy 简单了一些

region切分的阈值依然和待分裂region所属表在当前regionserver上的region个数有关系

  • 如果region个数等于1,切分阈值为flush size 128M * 2
  • 否则为MaxRegionFileSize
  • 这种切分策略对于大集群中的大表、小表会比IncreasingToUpperBoundRegionSplitPolicy 更加友好,小表不会再产生大量的小region,而是适可而止。
4、KeyPrefixRegionSplitPolicy

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

5、DelimitedKeyPrefixRegionSplitPolicy

保证相同前缀的数据在同一个region中,例如rowKey的格式为:userid_eventtype_eventid,指定的delimiter为_ ,则split的的时候会确保userid相同的数据在同一个region中。

6、DisabledRegionSplitPolicy

不启用自动拆分, 需要指定手动拆分(方便运维)

Views: 39

HBase(十一)RowKey设计和热点问题

RowKey设计三原则

1、长度原则

rowkey是一个二进制码流,可以是任意字符串,最大长度64kb,实际应用中一般为10-100bytes,以byte[]形式保存,一般设计成定长。

  • 建议尽可能短;但是也不能太短,否则rowkey前缀重复的概率增大
  • 设计过长会降低memstore内存的利用率和HFile存储数据的效率。

2、散列原则

  • 建议将rowkey的高位作为散列字段,这样将提高数据均衡分布在每个RegionServer,以实现负载均衡的几率。
  • 如果没有散列字段,首字段直接是时间信息。所有的数据都会集中在一个RegionServer上,这样在数据检索的时候负载会集中在个别的RegionServer上,造成热点问题,会降低查询效率。

3、唯一原则

必须在设计上保证其唯一性,rowkey是按照字典顺序排序存储的,因此设计rowkey的时候,要充分利用这个排序的特点,可以将经常读取的数据存储到一块,将最近可能会被访问的数据放到一块

HBase表的热点

检索habse的记录首先要通过row key来定位数据行。
当大量的client访问hbase集群的一个或少数几个节点,造成少数region server的读/写请求过多、负载过大,而其他region server负载却很小,就造成了“热点”现象。

热点的解决方案

1、预分区

预分区的目的让表的数据可以均衡的分散在集群中,而不是默认只有一个region分布在集群的一个节点上。

2、加盐

这里所说的加盐不是密码学中的加盐,而是在rowkey的前面增加随机数,具体就是给rowkey分配一个随机前缀以使得它和之前的rowkey的开头不同

3 哈希

哈希会使同一行永远用一个前缀加盐。哈希也可以使负载分散到整个集群,但是读却是可以预测的。使用确定的哈希可以让客户端重构完整的rowkey,可以使用get操作准确获取某一个行数据。

rowkey=MD5(username).subString(0,10)+时间戳

4、反转

反转固定长度或者数字格式的rowkey。这样可以使得rowkey中经常改变的部分(最没有意义的部分)放在前面。

这样可以有效的随机rowkey,但是牺牲了rowkey的有序性。

电信公司:
移动-----------> 136xxxx9301  ----->1039xxxx631
                136xxxx1234
                136xxxx2341
电信
联通

再看另一个例子,需求:后期想经常按照居住地和年龄进行查询,RowKey该如何设计?

user表
rowkey    name    age   sex    address
          lisi1    21     m       beijing
          lisi2    22     m       beijing
          lisi3    25     m       beijing
          lisi4    30     m       beijing
          lisi5    40     f       shanghai
          lisi6    50     f       tianjin

解决方案:

rowkey= address+age+随机数:

beijing21+随机数
beijing22+随机数
beijing25+随机数
beijing30+随机数

Views: 22

HBase(十)整合Hive

Hive的安装部署

先决条件

  • 搭建好三节点Hadoop集群;
  • hadoop100上先安装好MySQL服务;

准备安装包

  • 下载hive安装包 apache-hive-3.1.2-bin.tar.gz
  • 规划安装目录 /opt/pkg/hive
  • 上传安装包到hadoop100服务器中

解压

解压安装包到指定的规划目录/opt/pkg/

[hadoop@hadoop100 ~]$ cd /export/softwares/
[hadoop@hadoop100 soft]$ tar -xzvf apache-hive-3.1.2-bin.tar.gz -C /opt/pkg/

修改配置文件

  1. 进入hive安装目录
[hadoop@hadoop100 servers]$ cd /opt/pkg
  1. 重新命名hive目录
[hadoop@hadoop100 install]$ mv  apache-hive-3.1.2-bin/    hive
  1. 修改/opt/pkg//conf目录下的hive-site.xml, 默认没有该文件, 需要手动创建
[hadoop@hadoop100 install]$ cd /opt/pkg/hive/conf/
[hadoop@hadoop100 conf]$ vim hive-site.xml
  1. 进入编辑模式, 文件内容如下
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
      <property>
              <name>javax.jdo.option.ConnectionURL</name>
              <value>jdbc:mysql://hadoop100:3306/metastore?useSSL=false</value>
      </property>

      <property>
              <name>javax.jdo.option.ConnectionDriverName</name>
              <value>com.mysql.jdbc.Driver</value>
      </property>
      <property>
              <name>javax.jdo.option.ConnectionUserName</name>
              <value>root</value>
      </property>
          <property>
              <name>hive.metastore.warehouse.dir</name>
              <value>/user/hive/warehouse</value>
      </property>
      <property>
              <name>javax.jdo.option.ConnectionPassword</name>
              <value>niit1234</value>
      </property>
      <property>
              <name>hive.metastore.schema.verification</name>
              <value>false</value>
      </property>
      <property>
              <name>hive.metastore.event.db.notification.api.auth</name>
              <value>false</value>
      </property>
       <property>
              <name>hive.cli.print.current.db</name>
              <value>true</value>
      </property>
       <property>
              <name>hive.cli.print.header</name>
              <value>true</value>
      </property>
      <property>
              <name>hive.server2.thrift.bind.host</name>
              <value>hadoop100</value>
      </property>
      <property>
              <name>hive.server2.thrift.port</name>
              <value>10000</value>
      </property>
</configuration>
  1. 修改日志配置文件hive-log4j.properties

创建hive日志存储目录

[hadoop@hadoop100 conf]$ mkdir /opt/pkg/hive/logs/

重命名生成文件hive-log4j.properties

[hadoop@hadoop100 conf]$ pwd
/opt/pkg/hive/conf
[hadoop@hadoop100 conf]$ mv hive-log4j2.properties.template hive-log4j2.properties
[hadoop@hadoop100 conf]$ vim hive-log4j2.properties # 修改文件

修改此文件的hive.log.dir属性的值

  #更改以下内容,设置我们的hive的日志文件存放的路径,便于排查问题
  hive.log.dir=/opt/pkg/hive/logs/

拷贝mysql驱动包

上传mysql驱动包,如mysql-connector-java-5.1.38.jar/opt/download/目录中

由于运行hive时,需要向mysql数据库中读写元数据,所以需要将mysql的驱动包上传到hive的lib目录下

[hadoop@hadoop100 ~]$ cd /opt/download/
[hadoop@hadoop100 soft]$ cp mysql-connector-java-5.1.38.jar /opt/pkg/hive/lib/

解决日志Jar包冲突

# 进入lib目录
[hadoop@hadoop100 conf]$ cd /opt/pkg/hive/lib/
# 重新命名 或者直接删除
[hadoop@hadoop100 lib]$ mv log4j-slf4j-impl-2.10.0.jar log4j-slf4j-impl-2.10.0.jar.bak

配置Hive环境变量

打开/etc/profile文件

$ sudo vim /etc/profile

末尾添加如下内容

export HIVE_HOME=/opt/pkg/hive
export PATH=$PATH:$HIVE_HOME/bin

使修改生效

$ source /etc/profile

初始化元数据库

  1. 新建一个hadoop100连接, 登录MySQL
# 连接MySQL, 用户名root, 密码niit1234
[hadoop@hadoop100 ~]$ mysql -uroot -pniit1234
  1. 创建hive元数据, 需要和hive-site.xml中配置的一致
-- 创建数据库, 数据库名为: metastore
create database metastore;
show databases;
  1. 退出mysql
exit
  1. 初始化元数据库
  [hadoop@hadoop100 ~]$ schematool -initSchema -dbType mysql -verbose

看到schemaTool completed 表示初始化成功

验证安装

  • hadoop集群已启动
  • mysql服务已启动
  • 在hadoop100上任意目录启动hive cli命令行客户端
[hadoop@hadoop100 ~]$ hive
# 启动成功之后, 警告信息可以忽略

查看有哪些数据库

show databases;

说明hive安装成功

退出cli

quit;

HBase与Hive整合配置

拷贝jar包

将我们HBase的五个jar包拷贝到hive的lib目录下
hbase的jar包都在/opt/pkg/hbase/lib
我们需要拷贝五个jar包名字如下

hbase-client-2.2.2.jar
hbase-hadoop2-compat-2.2.2.jar
hbase-hadoop-compat-2.2.2.jar
hbase-it-2.2.2.jar
hbase-server-2.2.2.jar
  • 在hadoop100执行以下命令
cd  /opt/pkg/hbase-2.2.2/lib
cp hbase-protocol-2.2.2.jar   /opt/pkg/hive/lib/
cp hbase-server-2.2.2.jar  /opt/pkg/hive/lib/
cp hbase-client-2.2.2.jar  /opt/pkg/hive/lib/
cp hbase-common-2.2.2.jar   /opt/pkg/hive/lib/
cp hbase-common-2.2.2-tests.jar   /opt/pkg/hive/lib/

如果发现hive/lib已经存在相同jar包(大版本相同即可),也可以先不拷贝。

修改hive的配置文件

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

cd /opt/pkg/hive/conf
vim hive-site.xml

添加以下两个属性的配置

<property>
    <name>hive.zookeeper.quorum</name>
    <value>hadoop100,hadoop101,hadoop102</value>
</property>
<property>
    <name>hbase.zookeeper.quorum</name>
    <value>hadoop100,hadoop101,hadoop102</value>
</property>

修改hive-env.sh配置文件

cd /opt/pkg/apache-hive-3.1.2/conf
cp hive-env.sh.template hive-env.sh
vim hive-env.sh
  • 添加以下配置
export HADOOP_HOME=/opt/pkg/hadoop
export HBASE_HOME=/opt/pkg/hbase
export HIVE_CONF_DIR=/opt/pkg/hive/conf

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

hive当中建表

  • hadoop100执行以下命令,进入hive客户端,并创建hive表
cd /opt/pkg/hive/
bin/hive

创建hive数据库与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 ;

准备数据内容如下并加载到hive表

hadoop100执行以下命令,创建数据文件

cd /opt/data
vim hive-hbase.txt

文件内容如下

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

vim编辑时 :set noexpandtab关闭将tab转换成空格的设置,输入时使用制表符作为分隔符
可以:set list用来查看制表符(^I)和换行($)

进入hive客户端进行加载数据

hive (course)> load data local inpath '/opt/data/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" = "cf:name,cf:score") tblproperties("hbase.table.name" = "hbase_score");

通过insert overwrite select 插入数据

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

hbase当中查看表 hbase_score

进入hbase的客户端查看表hbase_score,并查看当中的数据

hbase(main):023:0> list
TABLE
hbase_score
myuser
myuser2
student
user
5 row(s) in 0.0210 seconds
=> ["hbase_score", "myuser", "myuser2", "student", "user"]

hbase(main):024:0> scan 'hbase_score'

ROW                      COLUMN+CELL                                                   
 1                       column=cf:name, timestamp=1550628395266, value=zhangsan       
 1                       column=cf:score, timestamp=1550628395266, value=80           
 2                       column=cf:name, timestamp=1550628395266, value=lisi           
 2                       column=cf:score, timestamp=1550628395266, value=60           
 3                       column=cf:name, timestamp=1550628395266, value=wangwu         
 3                       column=cf:score, timestamp=1550628395266, value=30           
 4                       column=cf:name, timestamp=1550628395266, value=zhaoliu       
 4                       column=cf:score, timestamp=1550628395266, value=70           
4 row(s) in 0.0360 seconds

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

HBase当中创建表并手动插入加载一些数据

进入HBase的shell客户端,

bin/hbase shell

手动创建一张表,并插入加载一些数据进去

# 创建一张表
# 通过put插入数据到hbase表
create 'hbase_hive_score',{ NAME =>'cf'}
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");

查看hive表course.hbase2hive

select * from course.hbase2hive;

Views: 31

HBase(九)集成MapReduce

HBase表中的数据最终都是存储在HDFS上,HBase天生的支持MR的操作,我们可以通过MR直接处理HBase表中的数据,并且MR可以将处理后的结果直接存储到HBase表中。

参考: http://hbase.apache.org/book.html#mapreduce

实战一:HBase表到HBase表

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

第一步:创建myuser2这张hbase表

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

create 'myuser2','f1'

第二步:创建maven工程并导入jar包

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>XZK</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>HbaseMrDdemo</artifactId>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-auth</artifactId>
            <version>3.1.4</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-client -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-mapreduce</artifactId>
            <version>2.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>2.2.2</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testng</groupId>
            <artifactId>testng</artifactId>
            <version>6.14.3</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                    <!--    <verbal>true</verbal>-->
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.2</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*/RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

第三步:开发MR程序实现功能

自定义map类

package com.kkb.hbase.mr;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;

import java.io.IOException;

public class HBaseReadMapper extends TableMapper<Text, Put> {
    /**
     * @param key     rowKey
     * @param value   rowKey此行的数据 Result类型
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        // 获得rowKey的字节数组
        byte[] rowKeyBytes = key.get();
        String rowKeyStr = Bytes.toString(rowKeyBytes);
        Text text = new Text(rowKeyStr);

        // 输出数据 -> 写数据 -> Put 构建Put对象
        Put put = new Put(rowKeyBytes);
        // 获取一行中所有的Cell对象
        Cell[] cells = value.rawCells();
        // 将f1 : name& age输出
        for (Cell cell : cells) {
            //当前cell是否是f1
            //列族
            byte[] familyBytes = CellUtil.cloneFamily(cell);
            String familyStr = Bytes.toString(familyBytes);
            if ("f1".equals(familyStr)) {
                //在判断是否是name | age
                byte[] qualifier_bytes = CellUtil.cloneQualifier(cell);
                String qualifierStr = Bytes.toString(qualifier_bytes);
                if ("name".equals(qualifierStr)) {
                    put.add(cell);
                }
                if ("age".equals(qualifierStr)) {
                    put.add(cell);
                }
            }
        }

        // 判断是否为空;不为空,才输出
        if (!put.isEmpty()) {
            context.write(text, put);
        }
    }
}

自定义reduce类

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.Text;
import java.io.IOException;

/**
 * TableReducer第三个泛型包含rowkey信息
 */
public class HBaseWriteReducer extends TableReducer<Text, Put, ImmutableBytesWritable> {
    //将map传输过来的数据,写入到hbase表
    @Override
    protected void reduce(Text key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
        //rowkey
        ImmutableBytesWritable immutableBytesWritable = new ImmutableBytesWritable();
        immutableBytesWritable.set(key.toString().getBytes());

        //遍历put对象,并输出
        for(Put put: values) {
            context.write(immutableBytesWritable, put);
        }
    }
}

main入口类

package com.kkb.hbase.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        Configuration configuration = HBaseConfiguration.create();
        // 设定绑定的zk集群
        configuration.set("hbase.zookeeper.quorum", "hadoop100:2181,hadoop101:2181,hadoop102:2181");

        int run = ToolRunner.run(configuration, new Main(), args);
        System.exit(run);
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf());
        job.setJarByClass(Main.class);

        // mapper
        TableMapReduceUtil.initTableMapperJob(TableName.valueOf("myuser"), new Scan(), HBaseReadMapper.class, Text.class, Put.class, job);
        // reducer
        TableMapReduceUtil.initTable ReducerJob("myuser2", HBaseWriteReducer.class, job);

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    }
}

实战二: HDFS到HBase表

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

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

在/opt/data目录,创建user.txt文件

cd /opt/data
vim user.txt

内容如下:

0007    zhangsan    18
0008    lisi    25
0009    wangwu  20

将文件上传到hdfs的路径下面去

hdfs dfs -mkdir -p /hbase/input
hdfs dfs -put /opt/pkg/user.txt /hbase/input/

代码开发

package com.kkb.hbase.hdfs2hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 java.io.IOException;

/**
 * 将HDFS上文件/hbase/input/user.txt数据,导入到HBase的myuser2表
 */
public class HDFS2HBase {
    public static class HdfsMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

        // 数据原样输出
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value, NullWritable.get());
        }
    }

    public static class HBaseReducer extends TableReducer<Text, NullWritable, ImmutableBytesWritable> {

        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            /**
             * key -> 一行数据
             * 样例数据:
             * 0007 zhangsan    18
             * 0008 lisi    25
             * 0009 wangwu  20
             */
            String[] split = key.toString().split("\t");

            Put put = new Put(Bytes.toBytes(split[0]));
            put.addColumn("f1".getBytes(), "name".getBytes(), split[1].getBytes());
            put.addColumn("f1".getBytes(), "age".getBytes(), split[2].getBytes());

            context.write(new ImmutableBytesWritable(Bytes.toBytes(split[0])), put);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        // 设定zk集群
        conf.set("hbase.zookeeper.quorum", "hadoop100:2181,hadoop101:2181,hadoop102:2181");
        Job job = Job.getInstance(conf);

        job.setJarByClass(HDFS2HBase.class);

        // 可省略
        // job.setInputFormatClass(TextInputFormat.class);
        // 输入文件路径
        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop100:8020/hbase/input"));

        job.setMapperClass(HdfsMapper.class);
        // map端的输出的key value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        // 指定输出到hbase的表名
        TableMapReduceUtil.initTableReducerJob("myuser2", HBaseReducer.class, job);

        // 设置reduce个数
        job.setNumReduceTasks(1);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

实战三:HDFS到HBase表(批量加载)

需求

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

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

知识点描述:

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

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

HBase数据正常写流程回顾

bulkload方式的处理示意图

好处

  • 导入过程不占用Region资源
  • 能快速导入海量的数据
  • 节省内存
  1. 开发生成HFile文件的代码

自定义map类

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

// 四个泛型中后两个,分别对应rowkey及put
public class BulkLoadMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split("\t");
        // 封装输出的rowkey类型
        ImmutableBytesWritable immutableBytesWritable = new ImmutableBytesWritable(split[0].getBytes());

        // 构建put对象
        Put put = new Put(split[0].getBytes());
        put.addColumn("f1".getBytes(), "name".getBytes(), split[1].getBytes());
        put.addColumn("f1".getBytes(), "age".getBytes(), split[2].getBytes());

        context.write(immutableBytesWritable, put);
    }
}

程序main

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
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.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class HBaseBulkLoad extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        Configuration configuration = HBaseConfiguration.create();
        //设定zk集群
        configuration.set("hbase.zookeeper.quorum", "hadoop100:2181,hadoop101:2181,hadoop102:2181");

        int run = ToolRunner.run(configuration, new HBaseBulkLoad(), args);
        System.exit(run);
    }
    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = super.getConf();
        Job job = Job.getInstance(conf);
        job.setJarByClass(HBaseBulkLoad.class);

        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop100:8020/hbase/input"));
        job.setMapperClass(BulkLoadMapper.class);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Put.class);

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("myuser2"));

        //使MR可以向myuser2表中,增量增加数据
        HFileOutputFormat2.configureIncrementalLoad(job, table, connection.getRegionLocator(TableName.valueOf("myuser2")));
        //数据写回到HDFS,写成HFile -> 所以指定输出格式为HFileOutputFormat2
        job.setOutputFormatClass(HFileOutputFormat2.class);
        HFileOutputFormat2.setOutputPath(job, new Path("hdfs://hadoop100:8020/hbase/out_hfile"));

        //开始执行
        boolean b = job.waitForCompletion(true);

        return b? 0: 1;
    }
}

3、观察HDFS上输出的结果

4、加载HFile文件到hbase表中

方式一:代码加载

  package kkk.hbase.demo3;

  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.hbase.HBaseConfiguration;
  import org.apache.hadoop.hbase.TableName;
  import org.apache.hadoop.hbase.client.Admin;
  import org.apache.hadoop.hbase.client.Connection;
  import org.apache.hadoop.hbase.client.ConnectionFactory;
  import org.apache.hadoop.hbase.client.Table;
  import org.apache.hadoop.hbase.tool.BulkLoadHFiles;

  public class LoadData {

      public static void main(String[] args) throws Exception {
          Configuration configuration = HBaseConfiguration.create();
          configuration.set("hbase.zookeeper.quorum", "hadoop100,hadoop101,hadoop102");
          // 获取数据库连接
          Connection connection = ConnectionFactory.createConnection(configuration);
          // 获取表的管理器对象
          Admin admin = connection.getAdmin();
          // 获取table对象
          TableName tableName = TableName.valueOf("myuser2");
          Table table = connection.getTable(tableName);
          // 构建BulkLoadHFiles加载HFile文件 hbase2.0 api
          BulkLoadHFiles load = BulkLoadHFiles.create(configuration);
          load.bulkLoad(tableName, new Path("hdfs://hadoop100:8020/hbase/out_hfile"));
      }

Views: 27

HBase(八)Region 合并操作

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

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

下面来看一下如何进行region合并:

通过Merge类冷合并Region

执行冷合并前,需要先关闭hbase集群

创建一张hbase表:

create 'test','info1',SPLITS => ['1000','2000','3000']

查看表region

file

需求:

通过查看UI界面,需要把test表中的前2个region数据进行合并,分别是:

  1. test,,1620148534743.8222cc22ab4acb18b726b3c6be1cb082.
  2. test,1000,1620148534743.b2e07e8922d0ffa6477ef2d7ff368bba.

这里通过org.apache.hadoop.hbase.util.Merge类来实现,不需要进入hbase shell,直接执行(需要先关闭hbase集群):

hbase org.apache.hadoop.hbase.util.Merge test test,,1620148534743.8222cc22ab4acb18b726b3c6be1cb082. test,1000,1620148534743.b2e07e8922d0ffa6477ef2d7ff368bba.

成功后可以界面观察分区变化,但是我这里运行找不到主类org.apache.hadoop.hbase.util.Merge。实际上冷合并不常用,因为需要停止集群损耗巨大,工作中更常使用热合并。

通过online_merge热合并Region

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

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

需求:需要把test表中的2个region数据进行合并:

  1. test,2000,1620148534743.68216f0389235352165d3b884db24b95.
  2. test,3000,1620148534743.c85c37595f74abb8eb0276c50bf700e6.

需要进入hbase shell:

hbase(main):001:0> merge_region '68216f0389235352165d3b884db24b95','c85c37595f74abb8eb0276c50bf700e6'

Took 11.3953 seconds

成功后观察界面

file

Views: 45