HBase Java API Guide Put

HBase Java API Guide Put

如果不是刻意需要单条写入的话，推荐使用Put List而不是写缓存，因为语义上更友好。锁管理采取隐式加锁，不建议主动加锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class PutExample {

    public static void main(String[] args) throws IOException {
        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());

        HTableInterface table = helper.getTable("testtable");
        Put put = new Put(Bytes.toBytes("row1"));
        put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"),
                Bytes.toBytes("val1"));
        put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"),
                Bytes.toBytes("val2"));
        table.put(put);
        table.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

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

public class PutListExample {

    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        HBaseHelper helper = HBaseHelper.getHelper(conf);
        HTableInterface table = helper.getTable("testtable");

        List<Put> puts = new ArrayList<Put>();
        Put put1 = new Put(Bytes.toBytes("row1"));
        put1.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val1"));
        puts.add(put1);
        Put put2 = new Put(Bytes.toBytes("row2"));
        put2.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val2"));
        puts.add(put2);
        Put put3 = new Put(Bytes.toBytes("row2"));
        put3.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"), Bytes.toBytes("val3"));
        puts.add(put3);
        table.put(puts);
        table.close();
    }
}

Put Write Buffer(autoFlush=false)

默认情况下，一次Put操作即要与Region Server执行一次RPC操作，其执行过程可以被拆分为以下三个部分：

T1：RTT(Round-Trip Time)，即网络往返时延，它指从客户端发送数据开始，到客户端收到来自服务端的确认，总共经历的时延，不包括数据传输的时间；

T2：数据传输时间，即Put所操作的数据在客户端与服务端之间传输所消耗的时间开销，当数据量大的时候，T2的时间开销不容忽略；

T3：服务端处理时间，对于Put操作，即写入WAL日志（如果设置了WAL标识为true）、更新MemStore等。

其中，T2和T3都是不可避免的时间开销，那么能不能减少T1呢？假设我们将多次Put操作打包起来一次性提交到服务端，则可以将T1部分的总时间从T1 * N降低为T1，其中T1指的是单次RTT时间，N为Put的记录条数。

正是出于上述考虑，HBase为用户提供了客户端缓存批量提交的方式（即Write Buffer）。假设RTT的时间较长，如1ms，则该种方式能够显著提高整个集群的写入性能。那么，什么场景下适用于该种模式呢？下面简单分析一下：

如果Put提交的是小数据（如KB级别甚至更小）记录，那么T2很小，因此，通过该种模式减少T1的开销，能够明显提高写入性能。

如果Put提交的是大数据（如MB级别）记录，那么T2可能已经远大于T1，此时T1与T2相比可以被忽略，因此，使用该种模式并不能得到很好的性能提升，不建议通过增大Write Buffer大小来使用该种模式。

Caution: 虽然写缓冲区的存在可以减少RPC调用，但是因为缓冲数据是存储在client端的，所以要权衡内存消耗。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class PutWriteBufferExample {

    public static void main(String[] args) throws IOException {
        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());
        HTableInterface table = helper.getTable("testtable");
        System.out.println("Auto flush: " + table.isAutoFlush());

        table.setAutoFlush(false);

        Put put1 = new Put(Bytes.toBytes("row1"));
        put1.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val1"));
        table.put(put1);

        Put put2 = new Put(Bytes.toBytes("row2"));
        put2.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val2"));
        table.put(put2);

        Put put3 = new Put(Bytes.toBytes("row3"));
        put3.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("val3"));
        table.put(put3);

        Get get = new Get(Bytes.toBytes("row1"));
        Result res1 = table.get(get);
        System.out.println("Result: " + res1);

        table.flushCommits();

        Result res2 = table.get(get);
        table.close();
        System.out.println("Result: " + res2);
    }
}

MultiThread Put with RowLock

若使用Put(byte[] row)来访问服务器，服务器会在调用期间隐式生产行锁，客户端是不能得到该锁的。若使用Put(byte[] row, RowLock rowLock)在客户端显式加锁。建议用户使用服务端隐式加锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RowLock;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class RowLockExample {

    private final static byte[] ROW1 = Bytes.toBytes("row1");
    private final static byte[] COLFAM1 = Bytes.toBytes("colfam1");
    private final static byte[] QUAL1 = Bytes.toBytes("qual1");
    private final static byte[] VAL1 = Bytes.toBytes("val1");
    private final static byte[] VAL2 = Bytes.toBytes("val2");
    private final static byte[] VAL3 = Bytes.toBytes("val3");

    private static Configuration conf = HBaseConfiguration.create();

    static class UnlockedPut implements Runnable {
        public void run() {
            try {
                HBaseHelper helper = HBaseHelper.getHelper(conf);
                HTableInterface table = helper.getTable("testtable");
                Put put = new Put(ROW1);
                put.add(COLFAM1, QUAL1, VAL3);
                long time = System.currentTimeMillis();
                System.out.println("Thread trying to put same row now...");
                table.put(put);
                table.close();
                System.out.println("Wait time: "
                        + (System.currentTimeMillis() - time) + "ms");
            } catch (IOException e) {
                System.err.println("Thread error: " + e);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        HBaseHelper helper = HBaseHelper.getHelper(conf);
        HTableInterface table = helper.getTable("testtable");

        System.out.println("Taking out lock...");
        RowLock lock = table.lockRow(ROW1);
        System.out.println("Lock ID: " + lock.getLockId());

        Thread thread = new Thread(new UnlockedPut());
        thread.start();

        try {
            System.out.println("Sleeping 5secs in main()...");
            Thread.sleep(5000);
        } catch (InterruptedException e) {
        }

        try {
            Put put1 = new Put(ROW1, lock);
            put1.add(COLFAM1, QUAL1, VAL1);
            table.put(put1);

            Put put2 = new Put(ROW1, lock);
            put2.add(COLFAM1, QUAL1, VAL2);
            table.put(put2);
        } catch (Exception e) {
            System.err.println("Error: " + e);
        } finally {
            System.out.println("Releasing lock...");
            table.unlockRow(lock);
            table.close();
        }
        try {
            thread.join();
        } catch (InterruptedException e) {
        }
        System.out.println("After thread ended...");
        helper.dump("testtable", new String[] { "row1" }, null, null);
    }
}

Delete

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class DeleteExample {

    public static void main(String[] args) throws IOException {
        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());
        HTableInterface table = helper.getTable("testtable");
        System.out.println("Before delete call...");
        helper.dump("testtable", new String[] { "row1" }, null, null);

        Delete delete = new Delete(Bytes.toBytes("row1"));

        delete.setTimestamp(1);

        delete.deleteColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"));
        delete.deleteColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual3"), 5);

        delete.deleteColumns(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"));
        delete.deleteColumns(Bytes.toBytes("colfam1"), Bytes.toBytes("qual3"),
                6);

        delete.deleteFamily(Bytes.toBytes("colfam1"));
        delete.deleteFamily(Bytes.toBytes("colfam1"), 3);

        table.delete(delete);

        table.close();
        System.out.println("After delete call...");
        helper.dump("testtable", new String[] { "row1" }, null, null);
    }
}

Delete List

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

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

public class DeleteListExample {

    public static void main(String[] args) throws IOException {
        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());
        HTableInterface table = helper.getTable("testtable");
        System.out.println("Before delete call...");
        helper.dump("testtable", new String[] { "row1", "row2", "row3" }, null, null);

        List<Delete> deletes = new ArrayList<Delete>();

        Delete delete1 = new Delete(Bytes.toBytes("row1"));
        delete1.setTimestamp(4);
        deletes.add(delete1);

        Delete delete2 = new Delete(Bytes.toBytes("row2"));
        delete2.deleteColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"));
        delete2.deleteColumns(Bytes.toBytes("colfam2"), Bytes.toBytes("qual3"), 5);
        deletes.add(delete2);

        Delete delete3 = new Delete(Bytes.toBytes("row3"));
        delete3.deleteFamily(Bytes.toBytes("colfam1"));
        delete3.deleteFamily(Bytes.toBytes("colfam2"), 3);
        deletes.add(delete3);

        table.delete(deletes);

        table.close();
        System.out.println("After delete call...");
        helper.dump("testtable", new String[] { "row1", "row2", "row3" }, null, null);
    }
}

Scan
HTable的getScanner方法，返回包含结果集的Result实例迭代器。Scan时，包括起始行但不包括终止行。通过addFamily来限制返回的列族，通过addColumn限制返回的列。通过setTimeRange限制时间范围，通过setTimestamp限制时间戳，通过setMaxVersions限制返回的版本。使用完ResultScanner之后应在try/cache中close掉。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class ScanExample {

    public static void main(String[] args) throws IOException {

        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());
        System.out.println("Adding rows to table...");
        helper.fillTable("testtable", 1, 100, 100, "colfam1", "colfam2");

        HTableInterface table = helper.getTable("testtable");

        System.out.println("Scanning table #1...");
        Scan scan1 = new Scan();
        ResultScanner scanner1 = table.getScanner(scan1);
        for (Result res : scanner1) {
            System.out.println(res);
        }
        scanner1.close();

        System.out.println("Scanning table #2...");
        Scan scan2 = new Scan();
        scan2.addFamily(Bytes.toBytes("colfam1"));
        ResultScanner scanner2 = table.getScanner(scan2);
        for (Result res : scanner2) {
            System.out.println(res);
        }
        scanner2.close();

        System.out.println("Scanning table #3...");
        Scan scan3 = new Scan();
        scan3.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("col-5"))
                .addColumn(Bytes.toBytes("colfam2"), Bytes.toBytes("col-33"))
                .setStartRow(Bytes.toBytes("row-10"))
                .setStopRow(Bytes.toBytes("row-20"));
        ResultScanner scanner3 = table.getScanner(scan3);
        for (Result res : scanner3) {
            System.out.println(res);
        }
        scanner3.close();

        table.close();
    }
}

Scan with caching and batch
扫描迭代器的next方法本质上和get请求一样是一次RPC请求，那么也和每次put小单元数据一样，会需要缓冲技术来提升性能。缓存分3个层面，全局(hbase-site.xml中的hbase.client.scanner.caching)、表级(HTable.setScannerCaching)和扫描器级(Scan.setCaching)，优先级依次增大。RPC请求次数和服务器及客户端的缓存大小是有折中的。若RPC请求越多，说明缓存越小，这样比较适合get大单元数据，因为在网络传输上所耗费的延迟比较大，而RPC请求的往返时间则可忽略不计；RPC请求越少，说明使用了大量缓存，这比较适合get小单元数据，但是这样每次next调用的时间会很长，而且返回的数据超出了 client端JVM堆大小，则会out of memory。而且当传输和处理数据时间超过配置的扫描器租约时间则会抛出ScannerTimeoutException。批处理技术，针对的是数据量非常大的行，甚至超过了客户端进程的内存容量。面向列，允许用户每次next返回若干列，例如在扫描器中设置setBatch(5)，则每次返回5列。灵活结合缓存和批处理技术，可以控制扫描器的RPC次数。RPC count = row count * column count / Min(column count, batch size) / scan caching size

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
package com.ali.pe3.hbase.demo.crud;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Appender;
import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.spi.LoggingEvent;

import com.ali.pe3.hbase.demo.util.HBaseHelper;

import java.io.IOException;

public class ScanCacheBatchExample {

    private static HTableInterface table = null;

    private static void scan(int caching, int batch) throws IOException {
        Logger log = Logger.getLogger("org.apache.hadoop");
        final int[] counters = { 0, 0 };
        Appender appender = new AppenderSkeleton() {
            @Override
            protected void append(LoggingEvent event) {
                String msg = event.getMessage().toString();
                if (msg != null && msg.contains("Call: next")) {
                    counters[0]++;
                }
            }

            public void close() {
            }

            public boolean requiresLayout() {
                return false;
            }
        };
        log.removeAllAppenders();
        log.setAdditivity(false);
        log.addAppender(appender);
        log.setLevel(Level.DEBUG);

        Scan scan = new Scan();
        scan.setCaching(caching);
        scan.setBatch(batch);
        ResultScanner scanner = table.getScanner(scan);
        for (@SuppressWarnings("unused")
        Result result : scanner) {
            counters[1]++;
        }
        scanner.close();
        System.out.println("Caching: " + caching + ", Batch: " + batch
                + ", Results: " + counters[1] + ", RPCs: " + counters[0]);
    }

    public static void main(String[] args) throws IOException {

        HBaseHelper helper = HBaseHelper.getHelper(HBaseConfiguration.create());
        helper.fillTable("testtable", 1, 10, 10, "colfam1", "colfam2");

        table = helper.getTable("testtable");

        scan(1, 1);
        scan(200, 1);
        scan(2000, 100);
        scan(2, 100);
        scan(2, 10);
        scan(5, 100);
        scan(5, 20);
        scan(10, 10);
    }
}
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true