Hbase java api

時間 2019-11-20

標籤 hbase java api 欄目 Hadoop 简体版

原文原文鏈接

本文中使用的是最原始的java api，沒有使用spring-data-hbase，只是使用spring管理Hbase的配置

查詢操做分爲以下幾個步驟：

1. 獲取Hbase配置，這裏的配置主要指hbase的地址。若是是Zookeeper管理的，可使用Zookeeper的地址和端口

2. 根據配置獲取Hbase鏈接：

connection = ConnectionFactory.createConnection(this.hbaseConfig.gethBaseConfiguration())

3. 根據Hbase鏈接，獲取HTable。

HTable hTable = (HTable) connection.getTable(TableName.valueOf(tableName));

4. 查詢。多行查詢（Scan方式）和單行查詢（Get方式）

示例代碼爲scan方式

scan.setStartRow(Bytes.toBytes(startRow));  scan.setStopRow(Bytes.toBytes(endRow));  scan.setCaching(4000);  scan.setBatch(3);  //設置列族 // scan.addFamily(Bytes.toBytes(columnFamily));  // 設置列  scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("createtime"));  scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("lat"));  scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("lon"));

5. 獲取查詢數據，以scan爲例

resultScanner = hTable.getScanner(scan);

6. 獲取Cell中的數據 java

for (Result result : resultScanner) {
    RwwData rwwData = new RwwData();  for (Cell cell : result.rawCells()) {
        String k = Bytes.toString(CellUtil.cloneQualifier(cell));  String v = Bytes.toString(CellUtil.cloneValue(cell));    } }

其中：

CellUtil.cloneQualifier(cell)

獲取的是每一列的名稱。

CellUtil.cloneValue(cell)

獲取的是每一列的名稱對應的值。

HbaseConfig spring

public class HbaseConfig {
    private static String zookeeperAddr = "";
    private static String zookeeperPoot = "2181";
    private static int zkRetry = 2;

    private static Configuration hBaseConfiguration = null;

    static {
        if (hBaseConfiguration == null) {
            Configuration configuration = new Configuration();
            configuration.set("hbase.zookeeper.quorum", zookeeperAddr);
            configuration.set("hbase.zookeeper.property.clientPort", zookeeperPoot);
            configuration.setInt("hbase.client.retries.number", zkRetry);
            hBaseConfiguration = HBaseConfiguration.create(configuration);
        }
    }

    public static Configuration gethBaseConfiguration() {
        return hBaseConfiguration;
    }

    public void setZookeeperAddr(String zookeeperAddr) {
        this.zookeeperAddr = zookeeperAddr;
    }

    public void setZookeeperPoot(String zookeeperPoot) {
        this.zookeeperPoot = zookeeperPoot;
    }

    public void setZkRetry(int zkRetry) {
        this.zkRetry = zkRetry;
    }
}

HbaseConnection api

public class HbaseConnection {

    private static Connection connection = null;
    private static HbaseConfig hbaseConfig;

    static {
        if (connection == null) {
            try {
                connection = ConnectionFactory.createConnection(hbaseConfig.gethBaseConfiguration());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static Connection getConnection() {
        return connection;
    }

    public void setHbaseConfig(HbaseConfig hbaseConfig) {
        this.hbaseConfig = hbaseConfig;
    }
}

HbaseService ide

public class HbaseService {

    private Logger logger = LoggerFactory.getLogger(HbaseService.class);
    private final long BASE = 2147483647;
    private HbaseConnection hbaseCcnnection;

    /**
     * @param tableName        : 表名
     * @param columnFamily：列族
     * @param phone：手機號
     * @param beginTime：查詢開始時間
     * @param endTime：查詢結束時間
     * @return
     * @throws IOException
     */
    public List<GeoPoint> getGeoPoints(final String tableName, final String columnFamily
            , String phone, long beginTime, long endTime)
            throws IOException, InvocationTargetException, IllegalAccessException, ParseException {
        List<GeoPoint> geoPointList = new ArrayList<>();
        Connection connection = HbaseConnection.getConnection();

        HTable hTable = (HTable) connection.getTable(TableName.valueOf(tableName));
        //構建scan
        Scan scan = new Scan();
        String startRow = this.getRowKey(phone, endTime);
        String endRow = this.getRowKey(phone, beginTime);

        System.out.println(startRow + " : " + endRow);
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(endRow));
        scan.setCaching(4000);
        scan.setBatch(3);
        //設置列族
//        scan.addFamily(Bytes.toBytes(columnFamily));
        // 設置列
        scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("createtime"));
        scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("lat"));
        scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("lon"));
        ResultScanner resultScanner = null;

        List<RwwData> rowDataList = new ArrayList<>();
        //獲取座標點
        try {
            resultScanner = hTable.getScanner(scan);
            for (Result result : resultScanner) {
                System.out.println("result: " + result);
                RwwData rwwData = new RwwData();
                System.out.println("rawCelles" + result.rawCells());
                for (Cell cell : result.rawCells()) {
                    String k = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String v = Bytes.toString(CellUtil.cloneValue(cell));
                    rwwData.set(k, v);

                }
                System.out.println(rwwData);
                rowDataList.add(rwwData);
            }
        } catch (Exception e) {
            logger.error("從Hbase中獲取座標點信息出錯", e);
            e.printStackTrace();
        } finally {
            resultScanner.close();
        }

        System.out.println("【rowDataList ： 】" + rowDataList.size());
        System.out.println(rowDataList);

        if (rowDataList.size() > 0) {
            //按照時間升序排序
            Collections.sort(rowDataList);
            System.out.println("【rowDataList sort ： 】" + rowDataList);
//            GeoPoint geoPoint = new GeoPoint();
            for (RwwData rwwData : rowDataList) {
                GeoPoint geoPoint = new GeoPoint();
//                BeanUtils.copyProperties(geoPoint, rwwData);
                geoPoint.x = rwwData.lat; //維度
                geoPoint.y = rwwData.lon; //經度
                geoPointList.add(geoPoint);

            }
        }
        System.out.println(geoPointList);
        return geoPointList;
    }

    /**
     * phone+(2147483647-指定時間秒數）
     *
     * @param phone
     * @param time
     * @return
     * @throws ParseException
     */
    private String getRowKey(String phone, long time) throws ParseException {
//        long phoneNum = NumberUtils.toLong(phone);
//        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        long timeMillSeconds = format.parse(time).getTime();
        long result = this.BASE - time / 1000;
        return phone + result;
    }

    /**
     *
     */
    private class RwwData implements Comparable<RwwData> {
        private String createtime;
        private double lat; //維度，lat
        private double lon; //經度，lon

        //可使用反射啊，用最簡單的
        public void set(String k, String value) throws NoSuchFieldException, IllegalAccessException {
            Preconditions.checkArgument(StringUtils.isNotBlank(k));
            Preconditions.checkArgument(value != null);
            Field f = RwwData.class.getDeclaredField(k);
            f.setAccessible(true);
            f.set(this, value);
//            if (StringUtils.equals("createtime", k)) {
//                this.createtime = String.valueOf(value);
//            } else if (StringUtils.equals("lat", k)) {
//                this.x = Double.parseDouble(value); //維度
//            } else if (StringUtils.equals("lon", k)) {
//                this.y = Double.parseDouble(value); //經度
//            } else {
//                throw new IllegalArgumentException(" no properties " + k + " found");
//            }

        }

        @Override
        public int compareTo(RwwData o) {
            return this.createtime.compareTo(o.createtime);
        }

        public double getLon() {
            return lon;
        }

        public void setLon(double lon) {
            this.lon = lon;
        }

        public double getLat() {
            return lat;
        }

        public void setLat(double lat) {
            this.lat = lat;
        }

        public String getCreatetime() {
            return createtime;
        }

        public void setCreatetime(String createtime) {
            this.createtime = createtime;
        }

        @Override
        public String toString() {
            return "RwwData{" +
                    "createtime='" + createtime + '\'' +
                    ", lat=" + lat +
                    ", lon=" + lon +
                    '}';
        }
    }

    public class GeoPoint {

        private double x;
        private double y;
        private double xyy;
        private double yx;
        private String fyx;

        public double getX() {
            return x;
        }

        public void setX(double x) {
            this.x = x;
        }

        public double getY() {
            return y;
        }

        public void setY(double y) {
            this.y = y;
        }

        public double getXyy() {
            return xyy;
        }

        public void setXyy(double xyy) {
            this.xyy = xyy;
        }

        public double getYx() {
            return yx;
        }

        public void setYx(double yx) {
            this.yx = yx;
        }

        public String getFyx() {
            return fyx;
        }

        public void setFyx(String fyx) {
            this.fyx = fyx;
        }

        @Override
        public String toString() {
            return "GeoPoint{" +
                    "x=" + x +
                    ", y=" + y +
                    '}';
        }
    }
}