Hadoop-Hbase

这篇具有很好参考价值的文章主要介绍了Hadoop-Hbase。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1. Hbase安装

1.1  安装zookeeper、 hbase

解压至/opt/soft,并分别改名

Hadoop-Hbase,hadoop,hbase,大数据

配置环境变量并source生效

#ZK

export ZOOKEEPER_HOME=/opt/soft/zk345

export PATH=$ZOOKEEPER_HOME/bin:$PATH

#HBASE_HOME

export HBASE_HOME=/opt/soft/hbase235

export PATH=$HBASE_HOME/bin:$PATH

hbase235/conf目录下的

编辑hbase-env.sh:[root@kb129 conf]# vim ./hbase-env.sh

export JAVA_HOME=/opt/soft/jdk180

Hadoop-Hbase,hadoop,hbase,大数据

Hadoop-Hbase,hadoop,hbase,大数据

编辑hbase-site.xml

[root@kb129 conf]# vim ./hbase-site.xml

<property>

    <name>hbase.rootdir</name>

    <value>hdfs://192.168.142.129:9000/hbase</value>

  </property>

  <property>

    <name>hbase.cluster.distributed</name>

    <value>true</value>

  </property>

  <property>

    <name>hbase.zookeeper.property.dataDir</name>

    <value>/opt/soft/zk345/zkdata</value>

  </property>

  <property>

    <name>hbase.zookeeper.property.clientPort</name>

    <value>2181</value>

  </property>

拷贝配置文件准备配置zookeeper

[root@kb129 conf]# pwd

/opt/soft/zk345/conf

[root@kb129 conf]# cp zoo_sample.cfg zoo.cfg

创建目录

[root@kb129 conf]# mkdir /opt/soft/zk345/logs

[root@kb129 conf]# mkdir /opt/soft/zk345/zkdata

Hadoop-Hbase,hadoop,hbase,大数据

编辑配置文件,增加目录指向

[root@kb129 conf]# vim  ./zoo.cfg

dataDir=/opt/soft/zk345/zkdata

dataLogDir=/opt/soft/zk345/logs

server.0=192.168.142.129:2287:3387

Hadoop-Hbase,hadoop,hbase,大数据

追加节点id

[root@kb129 conf]# cd ../zkdata/

[root@kb129 zkdata]# echo "0">myid

启动zookeeper

[root@kb129 zkdata]# zkServer.sh start

Hadoop-Hbase,hadoop,hbase,大数据

启动hbase

(启动前解决Hbase和hadoop中log4j的jar包冲突报错问题:将Hbase中的jar包改名就不会读取,解决掉冲突报错:mv /opt/soft/hbase235/lib/client-facing-thirdparty/slf4j-log4j12-1.7.30.jar /opt/soft/hbase235/lib/client-facing-thirdparty/slf4j-log4j12-1.7.30.jar.bak)

Hadoop-Hbase,hadoop,hbase,大数据

[root@kb129 zkdata]# start-hbase.sh

Hadoop-Hbase,hadoop,hbase,大数据

Hadoop-Hbase,hadoop,hbase,大数据

Hadoop-Hbase,hadoop,hbase,大数据

网页访问http://kb129:16010/

Hadoop-Hbase,hadoop,hbase,大数据

hbase宕机恢复操作(此操作会删除hbase中数据,慎用!!!)
(1)执行stop-hbase.sh关闭hbase进程,或通过kill杀死进程
(2)确保hadoop和zookeeper正常运行状态下,进入zookeeper客户端:zkCli.sh
(3)删除hbase:rmr /hbase,删除后:ls / ,查看是否已经删除
(4)进入hdfs系统,删除hbase指向目录/hbase
(5)执行start-hbase.sh,成功恢复hbase

2.Hbase shell相关操作

[root@kb129 conf]# hbase shell

创建命名空间

hbase(main):001:0> create_namespace 'kb23'

查看命名空间

hbase(main):001:0>list _namespace

Hadoop-Hbase,hadoop,hbase,大数据 

创建表

hbase(main):007:0> create 'bigdata:student','baseinfo','schoolinfo'

查看表

hbase(main):008:0> list_namespace_tables 'bigdata'

查看描述

hbase(main):009:0> desc 'bigdata:student'

删除表之前先禁用表

Hadoop-Hbase,hadoop,hbase,大数据

禁用表

hbase(main):011:0> disable 'bigdata:student'

查看是否禁用/启用

hbase(main):011:0> is_disabled/is_enabled 'bigdata:student'

启用表

hbase(main):011:0> enable 'bigdata:student'

插入数据(原有数据的话会覆盖)

hbase(main):006:0> put 'bigdata:student','rowkey1','baseinfo:name','tom'

查看指定rowkey的所有数据

hbase(main):008:0> get 'bigdata:student','rowkey1'

COLUMN                        CELL                                                                               

 baseinfo:age                 timestamp=2023-09-13T14:21:26.095, value=30                                        

 baseinfo:name                timestamp=2023-09-13T14:21:22.385, value=tom    

查看指定列族数据

hbase(main):011:0> get 'bigdata:student','rowkey1','baseinfo'

COLUMN                        CELL                                                                               

 baseinfo:age                 timestamp=2023-09-13T14:21:26.095, value=30                                        

 baseinfo:name                timestamp=2023-09-13T14:21:22.385, value=tom   

查看指定列族中某列数据

hbase(main):012:0> get 'bigdata:student','rowkey1','baseinfo:name'

COLUMN                        CELL                                                                               

 baseinfo:name                timestamp=2023-09-13T14:21:22.385, value=tom

添加列族

hbase(main):021:0> alter 'kb23:student','teacherinfo'

删除列族

hbase(main):023:0> alter 'kb23:student',{NAME=>'teacherinfo',METHOD=>'delete'}

更改版本个数(VERSIONS默认为1)

hbase(main):026:0> alter 'kb23:student',{NAME=>'baseinfo',VERSIONS=>3}

全表扫描

hbase(main):027:0> scan 'kb23:student'

删除指定列族中的列

hbase(main):055:0> delete 'kb23:student','rowkey2','baseinfo:name'

删除所有rowkey2信息

hbase(main):060:0> deleteall 'kb23:student','rowkey2'

查看不同版本信息

hbase(main):070:0> get 'kb23:student','rowkey1',COLUMN=>'baseinfo:name',VERSIONS=>3

查看范围内rowkey的数据(左闭右开)

hbase(main):094:0> scan 'kb23:student', {COLUMNS => 'baseinfo:name', STARTROW => 'rowkey',STOPROW=> 'rowkey3'}

查看时使用limit

hbase(main):096:0> scan 'kb23:student', {COLUMNS => 'baseinfo:name', STARTROW => 'rowkey',STOPROW=> 'rowkey3', VERSIONS=> 3, LIMIT=> 1}

过滤查找value包含11开头的信息(可查到多个版本)

hbase(main):099:0> scan 'kb23:student',FILTER=>"ValueFilter(=,'binary:11')"

过滤查找value包含andemen开头的信息(可查到多个版本)

hbase(main):102:0> scan 'kb23:student',FILTER=>"ValueFilter(=,'substring:andemen')"

过滤查找列名birth开头的信息(可查到多个版本)

hbase(main):104:0> scan 'kb23:student',FILTER=>"ColumnPrefixFilter('birth')"

多条件AND查询

hbase(main):110:0> scan 'kb23:student',FILTER=>"ColumnPrefixFilter('birth') AND ValueFilter(=,'substring:200')"

AND或OR查询

hbase(main):005:0>scan 'kb23:student',FILTER=>"ColumnPrefixFilter('birth') AND (ValueFilter(=,'substring:200')) OR ValueFilter(=,'substring:20')"

3.Hbase运行原理

3.1  HBase物理架构

Hadoop-Hbase,hadoop,hbase,大数据

1)StoreFile

保存实际数据的物理文件,StoreFile以Hfile的形式存储在HDFS上。每个Store会有一个或多个StoreFile(HFile),数据在每个StoreFile中都是有序的。

2)MemStore

写缓存,由于HFile中的数据要求是有序的,所以数据是先存储在MemStore中,排好序后,等到达刷写时机才会刷写到HFile,每次刷写都会形成一个新的HFile。

3)WAL

由于数据要经MemStore排序后才能刷写到HFile,但把数据保存在内存中会有很高的概率导致数据丢失,为了解决这个问题,数据会先写在一个叫做Write-Ahead logfile的文件中,然后再写入MemStore中。所以在系统出现故障的时候,数据可以通过这个日志文件重建。

4)BlockCache

读缓存,每次查询出的数据会缓存在BlockCache中,方便下次查询。

3.2  写流程

Hadoop-Hbase,hadoop,hbase,大数据

1)Client先访问zookeeper,获取hbase:meta表位于哪个Region Server。

2)访问对应的Region Server,获取hbase:meta表,根据写请求的namespace:table/rowkey,查询出目标数据位于哪个Region Server中的哪个Region中。并将该table的region信息以及meta表的位置信息缓存在客户端的meta cache,方便下次访问。

3)与目标Region Server进行通讯;

4)将数据顺序写入(追加)到WAL;

5)将数据写入对应的MemStore,数据会在MemStore进行排序;

6)向客户端发送ack;

7)等达到MemStore的刷写时机后,将数据刷写到HFile。

3.3  MemStore Flush

Hadoop-Hbase,hadoop,hbase,大数据

MemStore刷写时机:

1.当某个memstroe的大小达到了hbase.hregion.memstore.flush.size(默认值128M),其所在region的所有memstore都会刷写。

当memstore的大小达到了

hbase.hregion.memstore.flush.size(默认值128M)* hbase.hregion.memstore.block.multiplier(默认值4)时,会阻止继续往该memstore写数据。

2.当region server中memstore的总大小达到

java_heapsize*hbase.regionserver.global.memstore.size(默认值0.4)*hbase.regionserver.global.memstore.size.lower.limit(默认值0.95),

region会按照其所有memstore的大小顺序(由大到小)依次进行刷写。直到region server中所有memstore的总大小减小到上述值以下。

当region server中memstore的总大小达到

java_heapsize*hbase.regionserver.global.memstore.size(默认值0.4)时,会阻止继续往所有的memstore写数据。

3. 到达自动刷写的时间,也会触发memstore flush。自动刷新的时间间隔由该属性进行配置hbase.regionserver.optionalcacheflushinterval(默认1小时)。

4.当WAL文件的数量超过hbase.regionserver.max.logs,region会按照时间顺序依次进行刷写,直到WAL文件数量减小到hbase.regionserver.max.log以下(该属性名已经废弃,现无需手动设置,最大值为32)。

3.4  读流程

Hadoop-Hbase,hadoop,hbase,大数据

Hadoop-Hbase,hadoop,hbase,大数据

1)Client先访问zookeeper,获取hbase:meta表位于哪个Region Server。

2)访问对应的Region Server,获取hbase:meta表,根据读请求的namespace:table/rowkey,查询出目标数据位于哪个Region Server中的哪个Region中。并将该table的region信息以及meta表的位置信息缓存在客户端的meta cache,方便下次访问。

3)与目标Region Server进行通讯;

4)分别在MemStore和Store File(HFile)中查询目标数据,并将查到的所有数据进行合并。此处所有数据是指同一条数据的不同版本(time stamp)或者不同的类型(Put/Delete)。

5)将查询到的新的数据块(Block,HFile数据存储单元,默认大小为64KB)缓存到Block Cache。

6)将合并后的最终结果返回给客户端。

3.5  StoreFile Compaction

由于memstore每次刷写都会生成一个新的HFile,且同一个字段的不同版本(timestamp)和不同类型(Put/Delete)有可能会分布在不同的HFile中,因此查询时需要遍历所有的HFile。为了减少HFile的个数,以及清理掉过期和删除的数据,会进行StoreFile Compaction。

Compaction分为两种,分别是Minor Compaction和Major Compaction。Minor Compaction会将临近的若干个较小的HFile合并成一个较大的HFile,并清理掉部分过期和删除的数据。Major Compaction会将一个Store下的所有的HFile合并成一个大HFile,并且会清理掉所有过期和删除的数据。

Hadoop-Hbase,hadoop,hbase,大数据

3.6  Region Split

默认情况下,每个Table起初只有一个Region,随着数据的不断写入,Region会自动进行拆分。刚拆分时,两个子Region都位于当前的Region Server,但处于负载均衡的考虑,HMaster有可能会将某个Region转移给其他的Region Server。

Region Split时机:

1.当1个region中的某个Store下所有StoreFile的总大小超过hbase.hregion.max.filesize (10G),该Region就会进行拆分(0.94版本之前)。

2.当1个region中的某个Store下所有StoreFile的总大小超过Min(initialSize*R^3 ,hbase.hregion.max.filesize"),该Region就会进行拆分。其中initialSize的默认值为2*hbase.hregion.memstore.flush.size,R为当前Region Server中属于该Table的Region个数(0.94版本之后)。

具体的切分策略为:

第一次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了。

3.Hbase 2.0引入了新的split策略:如果当前RegionServer上该表只有一个Region,按照2 * hbase.hregion.memstore.flush.size分裂,否则按照hbase.hregion.max.filesize分裂。

Hadoop-Hbase,hadoop,hbase,大数据

4.hive映射hbase

hive中创建表格,关联hbase中的表

(1)

Hadoop-Hbase,hadoop,hbase,大数据

create external table student(
    id string,
    stuname string,
    birthdate string,
    age int,
    schoolname string,
    location string
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' with
    serdeproperties
('hbase.columns.mapping'=':key,baseinfo:name,baseinfo:birthday,baseinfo:age,schoolinfo:name,schoolinfo:location')
tblproperties ('hbase.table.name'='kb23:student');

select * from student;

Hadoop-Hbase,hadoop,hbase,大数据

(2)rowkey之间不同字段

Hadoop-Hbase,hadoop,hbase,大数据

create external table kb23factory(

    id string,

    usrname string,

    age string,

    gender string,

    email string,

    partname string,

    permission string

  )

  stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' with

    serdeproperties

        ('hbase.columns.mapping'=':key,usr:name,usr:age,usr:gender,usr:email,part:name,part:permission')

  tblproperties ('hbase.table.name'='kb23:kb23factory');

  

  select * from kb23factory;

Hadoop-Hbase,hadoop,hbase,大数据

5.Hbase API

5.1   pom依赖

<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-client</artifactId>
  <version>2.3.5</version>
</dependency>
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-common</artifactId>
  <version>2.3.5</version>
</dependency>

5.2  增删改查等具体操作

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

/**
 * Unit test for simple App.
 */
public class AppTest {

    private Connection connection = null;

    // 获取配置类
    private Configuration config = HBaseConfiguration.create();

    //创建连接
    @Before
    public void init() throws IOException {
        // 给配置类添加配置
        config.set("hbase.zookeeper.quorum", "kb129");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        // 获取连接
        connection = ConnectionFactory.createConnection(config);
    }

    //打印连接,测试连接
    @Test
    public void testConnection() throws IOException {
        System.out.println(connection);
    }

    //创建表
    @Test
    public void createTable() throws IOException {
        // 获取admin
        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf("kb23:test2");
        /*HTableDescriptor desc = new HTableDescriptor(tableName);
        HColumnDescriptor family1 = new HColumnDescriptor("info");
        HColumnDescriptor family2 = new HColumnDescriptor("info2");
        desc.addFamily(family1);
        desc.addFamily(family2);*/

        // 获取descriptor的builder
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // 添加列族
        ColumnFamilyDescriptor columnFamilyDescriptor1 = ColumnFamilyDescriptorBuilder.of("baseinfo");
        ColumnFamilyDescriptor columnFamilyDescriptor2 = ColumnFamilyDescriptorBuilder.of("schoolinfo");

        // 将单个列族的descriptor添加到builder中
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor1);
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor2);
        TableDescriptor descriptor = tableDescriptorBuilder.build();

        //创建表
        admin.createTable(descriptor);

    }

    //插入数据
    @Test
    public void putValue() throws IOException {
        // 1.获取table
        Table table = connection.getTable(TableName.valueOf("kb23:test1"));
        // 2.创建Put对象
        Put put = new Put(Bytes.toBytes("liuxin"));
        // 3.添加put属性
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("刘鑫"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Male"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("180"));
        // 4.put数据
        table.put(put);
        // 5.关闭资源
        table.close();
    }

    //通过集合插入多条数据
    @Test
    public void putValueList() throws IOException {
        // 1.获取table
        Table table = connection.getTable(TableName.valueOf("kb23:test1"));
        // 2.创建Put对象
        Put zs = new Put(Bytes.toBytes("zhangsan"));
        // 3.添加put属性
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("张三"));
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Female"));
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("160"));

        Put ls = new Put(Bytes.toBytes("lisi"));
        // 3.添加put属性
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("李四"));
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Male"));
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("180"));
        // 4.put数据
        ArrayList<Put> puts = new ArrayList<Put>();
        puts.add(zs);
        puts.add(ls);
        table.put(puts);
        // 5.关闭资源
        table.close();
    }


    @Test
    public void putValueMutator() throws IOException {
        BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf("kb23:test1 "));
        bufferedMutatorParams.setWriteBufferPeriodicFlushTimeoutMs(10000);//设置超时flush时间最大值
        bufferedMutatorParams.writeBufferSize(10*1024*1024);//设置缓存大小flush
        BufferedMutator bufferedMutator = connection.getBufferedMutator(bufferedMutatorParams) ;
        /*// 3.添加put属性
        Put zs = new Put(Bytes.toBytes("zhangsan"));
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("张三"));
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Female"));
        zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("160"));

        Put ls = new Put(Bytes.toBytes("lisi"));
        // 3.添加put属性
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("李四"));
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Male"));
        ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("180"));*/
        // 4.put数据
        ArrayList<Put> puts = new ArrayList<Put>();
        //puts.add(zs);
        //puts.add(ls);
        bufferedMutator.mutate(puts);
    }

    //删除数据
    @Test
    public void delete() throws IOException {
        // 1.获取table
        Table table = connection.getTable(TableName.valueOf("kb23:test1"));
        //删除具体的列
        /*Delete delname = new Delete(Bytes.toBytes("lisi"));
        delname.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"));
        table.delete(delname);*/

        //通过rowkey整个删除
        Delete lisi = new Delete(Bytes.toBytes("lisi"));
        table.delete(lisi);
    }

    //查询数据
    @Test
    public void getValue() throws IOException {
        // 1.获取table
        Table table = connection.getTable(TableName.valueOf("kb23:test1"));

        // 2.获取Get对象
        Get zhangsan = new Get(Bytes.toBytes("zhangsan"));

        // 3. 获取result
        Result result = table.get(zhangsan);
        byte[] name = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"));
        byte[] gender = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender"));
        byte[] height = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("height"));
        System.out.println(Bytes.toString(name) + Bytes.toString(gender) + Bytes.toString(height));
    }

    //扫描数据
    @Test
    public void scanValue() throws IOException {
        // 1.获取table
        Table table = connection.getTable(TableName.valueOf("kb23:test1"));

        // 2.创建Scan对象
        Scan scan = new Scan();

        // 3.扫描数据
        scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));
        scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"));
        scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"));
        ResultScanner scanner = table.getScanner(scan);

        // 4.获取结果
        for (Result result : scanner) {
            byte[] name = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"));
            byte[] gender = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender"));
            byte[] height = result.getValue(Bytes.toBytes("info"), Bytes.toBytes("height"));
            System.out.println(Bytes.toString(name) + Bytes.toString(gender) + Bytes.toString(height));
        }
    }

    //过滤器,查找数据
    @Test
    public void getFilterRowKey() throws IOException {
        TableName tableName = TableName.valueOf("kb23:student");
        Table table = connection.getTable(tableName);
        FamilyFilter familyFilter =
                new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("baseinfo")));
        Scan scan = new Scan();
        scan.setFilter(familyFilter);
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            byte[] name = result.getValue(Bytes.toBytes("baseinfo"), Bytes.toBytes("name"));
            byte[] age = result.getValue(Bytes.toBytes("baseinfo"), Bytes.toBytes("age"));
            byte[] birthday = result.getValue(Bytes.toBytes("baseinfo"), Bytes.toBytes("birthday"));
            System.out.println(Bytes.toString(name) + "\t" + Bytes.toString(age) + "\t" + Bytes.toString(birthday));
            System.out.println("----------------------");
        }
    }
   
    //关闭连接
    @After
    public void closeConnection() throws IOException {
        if (connection != null) {
            // 关闭连接
            connection.close();
        }
    }
}

5.3  写缓存  HBase客户端的批量写缓存BufferedMutator

HBase的每一个put操作实际上是一个RPC操作,将客户端的数据传输到服务器再返回结果,这只适用于小数据量的操作,如果数据量多的话,每次put都需要建立一次RPC的连接(TCP连接),而建立连接传输数据是需要时间的,因此减少RPC的调用可以提高数据传输的效率,减少建立连接的时间和IO消耗。

HBase的客户端API提供了写缓存区,put的数据一开始放在缓存区内,当数量到达指定的容量或者用户强制提交是才将数据一次性提交到HBase的服务器。这个缓冲区可以通过调用 HTable.setAutoFlush(false) 来开启。而新版HBbase的API中使用了BufferedMutator替换了老版的缓冲区,通过BufferedMutator对象提交的数据自动存放在缓冲区中。

BufferedMutator

通过获取 BufferedMutator 对象,并调用 mutator.mutate(List<Mutation> mutations) 方法来进行批量插入数据。可以使用 Put 类型的对象列表作为 mutations 参数进行插入。BufferedMutator 提供了自动管理缓冲区和写入操作的功能,可以提高插入数据的性能。文章来源地址https://www.toymoban.com/news/detail-733115.html

@Test
public void putValueMutator() throws IOException {
    BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf("kb23:test1 "));
    bufferedMutatorParams.setWriteBufferPeriodicFlushTimeoutMs(10000);//设置超时flush时间最大值
    bufferedMutatorParams.writeBufferSize(10*1024*1024);//设置缓存大小flush
    BufferedMutator bufferedMutator = connection.getBufferedMutator(bufferedMutatorParams) ;
    /*// 3.添加put属性
    Put zs = new Put(Bytes.toBytes("zhangsan"));
    zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("张三"));
    zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Female"));
    zs.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("160"));

    Put ls = new Put(Bytes.toBytes("lisi"));
    // 3.添加put属性
    ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("李四"));
    ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("Male"));
    ls.addColumn(Bytes.toBytes("info"), Bytes.toBytes("height"), Bytes.toBytes("180"));*/
    // 4.put数据
    ArrayList<Put> puts = new ArrayList<Put>();
    //puts.add(zs);
    //puts.add(ls);
    bufferedMutator.mutate(puts);
}

到了这里,关于Hadoop-Hbase的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Linux CentOS下大数据环境搭建(zookeeper+hadoop+hbase+spark+scala)

    本篇文章是结合我个人学习经历所写,如果遇到什么问题或者我有什么错误,欢迎讨论。 百度网盘链接:https://pan.baidu.com/s/1DCkQQVYqYHYtPws9hWGpgw?pwd=zh1y 提取码:zh1y 软件在连接中VMwareWorkstation_V16.2.1_XiTongZhiJia的文件夹下。 双击运行安装包,这里下一步即可。 这里勾选我接受许可

    2024年04月15日
    浏览(66)
  • 大数据系统常用组件理解(Hadoop/hive/kafka/Flink/Spark/Hbase/ES)

    一.Hadoop Hadoop是一个由Apache基金会所开发的分布式系统基础架构。 Hadoop 以一种可靠、高效、可伸缩的方式进行数据处理。 Hadoop的核心是yarn、HDFS和Mapreduce。yarn是资源管理系统,实现资源调度,yarn是Hadoop2.0中的资源管理系统,总体上是master/slave结构。对于yarn可以粗浅将其理解

    2024年02月20日
    浏览(44)
  • 轻大21级软工大数据实验(手把手教你入门Hadoop、hbase、spark)

    写在最前面,如果你只是来找答案的,那么很遗憾,本文尽量避免给出最后结果,本文适合Linux0基础学生,给出详细的环境配置过程,实验本身其实很简单,供大家一起学习交流。 1 .编程实现以下指定功能,并利用 Hadoop 提供的 Shell 命令完成相同任务 : 向HDFS 中上传任意文

    2024年02月05日
    浏览(83)
  • 大数据平台安装实验: ZooKeeper、Kafka、Hadoop、Hbase、Hive、Scala、Spark、Storm

    ​ 在大数据时代,存在很多开源的分布式数据采集、计算、存储技术,本实验将在熟练掌握几种常见Linux命令的基础上搭建几种常用的大数据采集、处理分析技术环境。 相关安装包下载: 链接:https://pan.baidu.com/s/1Wa2U3qstc54IAUCypcApSQ 提取码:lcd8 Hadoop大数据平台所需工具、软件

    2023年04月09日
    浏览(88)
  • Hadoop-Hbase

    1. Hbase安装 1.1  安装zookeeper、 hbase 解压至/opt/soft,并分别改名 配置环境变量并source生效 #ZK export ZOOKEEPER_HOME=/opt/soft/zk345 export PATH=$ZOOKEEPER_HOME/bin:$PATH #HBASE_HOME export HBASE_HOME=/opt/soft/hbase235 export PATH=$HBASE_HOME/bin:$PATH hbase235/conf 目录下的 编辑hbase-env.sh:[root@kb129 conf]# vim ./hbase-e

    2024年02月07日
    浏览(58)
  • Hadoop安装配置HBase

    1、文件准备 http://archive.apache.org/dist/hbase/1.3.1/下载hbase-1.3.1-bin.tar.gz并上传到虚拟机/root目录下。 2、安装          切换目录 [root@hadoop00 ~]# cd /usr/local 新建文件夹 [root@hadoop00 local]# mkdir hbase 解压文件 [root@hadoop00 local]# tar zxvf /root/hbase-1.3.1-bin.tar.gz -C /usr/local/hbase 3、环境变量

    2024年01月24日
    浏览(50)
  • hadoop -- Hbase

    HBase是一个分布式、可扩展、面向列的数据存储(百万级别列)、可伸缩、高可靠性、实时读写的NoSQL 数据库。 HBase利用 Hadoop的 HDFS作为其文件存储系统, 利用MapReduce 来处理HBase中的海量数据, 利用Zookeeper作为分布式协同服务。 HBase 基本的操作命令:  创建表:  创建一个

    2024年02月15日
    浏览(38)
  • HBase架构篇 - Hadoop家族的天之骄子HBase

    HBase 的数据存储在表中。表名是一个字符串。表由行和列组成。 HBase 的行由行键(rowkey)和 n 个列(column)组成。行键没有数据类型,可以看作是字节数组,类似于关系型数据库的主键索引,在整个 HBase 表中是唯一的,按照字母顺序排序。 HBase 的列族由多个列组成,相当于

    2023年04月20日
    浏览(34)
  • 实操Hadoop大数据高可用集群搭建(hadoop3.1.3+zookeeper3.5.7+hbase3.1.3+kafka2.12)

    前言 纯实操,无理论,本文是给公司搭建测试环境时记录的,已经按照这一套搭了四五遍大数据集群了,目前使用还未发现问题。 有问题麻烦指出,万分感谢! PS:Centos7.9、Rocky9.1可用 集群配置 ip hostname 系统 CPU 内存 系统盘 数据盘 备注 192.168.22.221 hadoop1 Centos7.9 4 16 250G 19

    2024年02月03日
    浏览(38)
  • 大数据期资料2023 Beta版 - Hadoop、HDFS、MapReduce、Hive、ZooKeeper、Kafka、HBase详解

    了解大数据概念、Hadoop、HDFS、MapReduce、Hive、ZooKeeper、Kafka、HBase等技术,包括特点、命令操作和启动关闭方法。获取2023年大数据资料Beta版。

    2024年02月06日
    浏览(233)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包