3

我在 bigtable 中有一个包含 10 个实例的小表（大小为 100Mb）。当我尝试每 1 分钟扫描/获取一行时，调用的延迟超过 300 毫秒。如果我打的是更频繁的呼叫，例如每秒一次，则延迟为 50-60 毫秒。我不确定如何通过低频调用提高性能。这是预期的行为。还是我做错了什么。

这是我的测试代码。我为两个与大表的 hbase 客户端连接创建了一个执行程序。但是低频连接响应比拨打更频繁呼叫的连接要慢得多。

有什么建议么？

package com.bids;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellUtil;
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.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.fusesource.jansi.AnsiConsole;

public class BTConnectTest {
    public static void main(String[] args) throws IOException, InterruptedException {

        Configuration hBaseConfig = HBaseConfiguration.create();
        hBaseConfig.set("google.bigtable.project.id", "xxxxxxx");
        hBaseConfig.set("google.bigtable.cluster.name", "hbase-test1");
        hBaseConfig.set("google.bigtable.zone.name", "us-central1-b");
        hBaseConfig.set("hbase.client.connection.impl", "com.google.cloud.bigtable.hbase1_1.BigtableConnection");

        ExecutorService executor = Executors.newSingleThreadExecutor();

        final Connection bigTableConnection1 = ConnectionFactory.createConnection(hBaseConfig, executor);

        final Connection bigTableConnection2 = ConnectionFactory.createConnection(hBaseConfig, executor);

        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e1) {
                        // TODO Auto-generated catch block
                        e1.printStackTrace();
                    }
                    long before = System.nanoTime();
                    try {
                        makeACall2Bigtable(bigTableConnection2);
                    } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    // bigTableConnection.close();
                    long after = System.nanoTime();

                    long diff = after - before;

                    System.out.println("\t\t\t\t\t\t connection: " + 1 + " diff: " + diff / (1000 * 1000));
                }
            }
        });
        t.start();

        long sum = 0;
        int n = 0;
        while (true) {
            if (n > 60) {
                Thread.sleep(60000);
            }

            long before = System.nanoTime();

            Connection bigTableConnection = bigTableConnection1;
            int label = -1;

            makeACall2Bigtable(bigTableConnection);
            long after = System.nanoTime();

            long diff = after - before;
            n = n + 1;
            sum += diff;
            long avg = sum / (n * 1000 * 1000);
            AnsiConsole a = new AnsiConsole();

            System.out.println("connection: " + 0 + " diff: " + diff / (1000 * 1000) + " avg: " + avg);

        }
        // bigTableConnection.close();

    }

    private static void makeACall2Bigtable(Connection bigTableConnection) throws IOException {

        Table table = bigTableConnection.getTable(TableName.valueOf("customer"));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes("101"));
        scan.setStopRow(Bytes.toBytes("102"));
        List<String> cols = new ArrayList<String>(3);
        cols.add("name");
        cols.add("age");
        cols.add("weight");
        String keyName = "id";
        final String DEFAULT_COLFAM = "z";
        for (String col : cols) {
            scan.addColumn(Bytes.toBytes(DEFAULT_COLFAM), Bytes.toBytes(col));
        }
        ResultScanner resultScanner = table.getScanner(scan);

        for (Result result : resultScanner) {
            Map<String, String> columnValueMap = new LinkedHashMap<String, String>();
            for (String col : cols) {
                if (result.containsColumn(Bytes.toBytes(DEFAULT_COLFAM), Bytes.toBytes(col))) {
                    columnValueMap.put(col, new String(CellUtil.cloneValue(
                            result.getColumnLatestCell(Bytes.toBytes(DEFAULT_COLFAM), Bytes.toBytes(col)))));
                } else {
                    if (cols.contains(keyName)) {
                        columnValueMap.put(col, null);
                    }

                }
            }

        }
        resultScanner.close();
        table.close();

    }

}
4

2 回答 2

5
  • 由于已知问题，前几个调用速度较慢。这是在服务器端为每个“频道”发生的一些设置，我们有多个频道。
  • 你不应该需要 finalFilterList。
  • 您应该缓存您的 Scan、TableName 和列族字节。您可以重复使用它们。
  • 如果您得到单行，请执行 Get 而不是扫描。
  • 需要执行人吗？
  • 为了安全起见，您的扫描可能应该使用 setMaxVersions(1)。
  • 也许尝试 scan.setStartRow(Bytes.toBytes("101")) 和 scan.setStopRow(Bytes.toBytes("102")) 而不是行前缀，看看是否有帮助？
  • 确保您的代码与集群在同一区域中运行。

我希望这会有所帮助。

于 2015-12-17T23:21:52.347 回答
3

如果您真的要在生产环境中执行低频请求，您可能希望运行一个后台线程，每隔几秒对您的表发出一个随机请求。

Bigtable 确实针对大量频繁访问的数据进行了优化。一段时间内的第一个请求可能会要求再次读入数据。定期请求将使其保持活动状态。

于 2015-12-18T00:02:50.973 回答