HBase 存储船舶轨迹方案

背景：项目需求要存储三个月的船舶轨迹数据，并且能够根据时间段快速查询单艘船的轨迹数据。船舶数据4~5万，5分钟记录一次轨迹。

1、环境基础：已经安装部署完成hbase 集群环境，安装部署链接: https://blog.csdn.net/weixin_41352552/article/details/109851768.

2、运用 springboot 集成hbase，使用定时器从redis中循环取出船舶点插入hbase,轨迹数量达到1千万。

1、目录架构

2、引入相关依赖 pom.xml 文件

<dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.3.0</version><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion><exclusion><groupId>javax.servlet</groupId><artifactId>servlet-api</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>2.6.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-core</artifactId><version>2.6.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-common</artifactId><version>2.6.0</version></dependency><dependency><groupId>org.springframework.data</groupId><artifactId>spring-data-hadoop</artifactId><version>2.5.0.RELEASE</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>2.6.0</version></dependency>

3、hbase连接配置

在application.properties 文件中添加以下配置

################################HBASE 配置##########################################hbase.zookeeper.quorum=master:2181,slave1:2181 # hbase集群的ip+端口

HBaseConn 连接配置

@Component
public class HBaseConn {@Value("${hbase.zookeeper.quorum}")private String zookeeperQuorum;private static final HBaseConn INSTANCE = new HBaseConn();private static Configuration configuration;private static Connection connection;private HBaseConn() {try {if (configuration == null) {configuration = HBaseConfiguration.create();configuration.set("hbase.zookeeper.quorum", zookeeperQuorum);}} catch (Exception e) {e.printStackTrace();}}private Connection getConnection() {if (connection == null || connection.isClosed()) {try {connection = ConnectionFactory.createConnection(configuration);} catch (Exception e) {e.printStackTrace();}}return connection;}public static Connection getHBaseConn() {return INSTANCE.getConnection();}public static Configuration getConfiguration() {return configuration;}public static void closeConn() {if (connection != null) {try {connection.close();} catch (IOException ioe) {ioe.printStackTrace();}}}public static Table getTable(String tableName) throws IOException {return INSTANCE.getConnection().getTable(TableName.valueOf(tableName));}
}

HBaseUtil Hbase的工具包

public class HBaseUtil {/*** 创建HBase表.** @param tableName 表名* @param cfs       列族的数组* @return 是否创建成功 boolean*/public static boolean createTable(String tableName, String[] cfs) {try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {if (admin.tableExists(tableName)) {return false;}HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));String nameAsString = tableDescriptor.getNameAsString();System.out.println(nameAsString);Arrays.stream(cfs).forEach(cf -> {HColumnDescriptor columnDescriptor = new HColumnDescriptor(cf);columnDescriptor.setMaxVersions(1);tableDescriptor.addFamily(columnDescriptor);});admin.createTable(tableDescriptor);} catch (Exception e) {e.printStackTrace();}return true;}/*** 删除hbase表.** @param tableName 表名* @return 是否删除成功 boolean*/public static boolean deleteTable(String tableName) {try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {admin.disableTable(tableName);admin.deleteTable(tableName);} catch (Exception e) {e.printStackTrace();}return true;}/*** hbase插入一条数据.** @param tableName 表名* @param rowKey    唯一标识* @param cfName    列族名* @param qualifier 列标识* @param data      数据* @return 是否插入成功 boolean*/public static boolean putRow(String tableName, String rowKey, String cfName, String qualifier,String data) {try (Table table = HBaseConn.getTable(tableName)) {Put put = new Put(Bytes.toBytes(rowKey));put.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier), Bytes.toBytes(data));table.put(put);} catch (IOException ioe) {ioe.printStackTrace();return false;}return true;}/*** 插入多条数据** @param tableName the table name* @param puts      the puts* @return the boolean*/public static boolean putRows(String tableName, List<Put> puts) {try (Table table = HBaseConn.getTable(tableName)) {table.put(puts);} catch (IOException ioe) {ioe.printStackTrace();return false;}return true;}/*** 获取单条数据.** @param tableName 表名* @param rowKey    唯一标识* @return 查询结果 row*/public static Result getRow(String tableName, String rowKey) {try (Table table = HBaseConn.getTable(tableName)) {Get get = new Get(Bytes.toBytes(rowKey));return table.get(get);} catch (IOException ioe) {ioe.printStackTrace();return null;}}/*** 获取数据** @param tableName  the table name* @param rowKey     the row key* @param filterList the filter list* @return the row*/public static Result getRow(String tableName, String rowKey, FilterList filterList) {try (Table table = HBaseConn.getTable(tableName)) {Get get = new Get(Bytes.toBytes(rowKey));get.setFilter(filterList);return table.get(get);} catch (IOException ioe) {ioe.printStackTrace();return null;}}/*** 检索数据..** @param tableName the table name* @return the scanner*/public static ResultScanner getScanner(String tableName) {try (Table table = HBaseConn.getTable(tableName)) {Scan scan = new Scan();scan.setCaching(1000);return table.getScanner(scan);} catch (IOException ioe) {ioe.printStackTrace();}return null;}/*** 批量检索数据.** @param tableName   表名* @param startRowKey 起始RowKey* @param endRowKey   终止RowKey* @return ResultScanner实例 scanner*/public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey) {try (Table table = HBaseConn.getTable(tableName)) {Scan scan = new Scan();scan.setStartRow(Bytes.toBytes(startRowKey));scan.setStopRow(Bytes.toBytes(endRowKey));scan.setCaching(1000);return table.getScanner(scan);} catch (IOException ioe) {ioe.printStackTrace();}return null;}/*** Gets scanner.** @param tableName   the table name* @param startRowKey the start row key* @param endRowKey   the end row key* @param filterList  the filter list* @return the scanner*/public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,FilterList filterList) {try (Table table = HBaseConn.getTable(tableName)) {Scan scan = new Scan();scan.setStartRow(Bytes.toBytes(startRowKey));scan.setStopRow(Bytes.toBytes(endRowKey));scan.setFilter(filterList);scan.setCaching(1000);return table.getScanner(scan);} catch (IOException ioe) {ioe.printStackTrace();}return null;}/*** HBase删除一行记录.** @param tableName 表名* @param rowKey    唯一标识* @return 是否删除成功 boolean*/public static boolean deleteRow(String tableName, String rowKey) {try (Table table = HBaseConn.getTable(tableName)) {Delete delete = new Delete(Bytes.toBytes(rowKey));table.delete(delete);} catch (IOException ioe) {ioe.printStackTrace();}return true;}/*** Delete column family boolean.** @param tableName the table name* @param cfName    the cf name* @return the boolean*/public static boolean deleteColumnFamily(String tableName, String cfName) {try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {admin.deleteColumn(tableName, cfName);} catch (Exception e) {e.printStackTrace();}return true;}/*** Delete qualifier boolean.** @param tableName the table name* @param rowKey    the row key* @param cfName    the cf name* @param qualifier the qualifier* @return the boolean*/public static boolean deleteQualifier(String tableName, String rowKey, String cfName,String qualifier) {try (Table table = HBaseConn.getTable(tableName)) {Delete delete = new Delete(Bytes.toBytes(rowKey));delete.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier));table.delete(delete);} catch (IOException ioe) {ioe.printStackTrace();}return true;}/*** 利用协处理器进行全表count统计** @param tableName the tableName* @return the long*/public static Long countRowsWithCoprocessor(String tableName) {try (Table table = HBaseConn.getTable(tableName);Connection hBaseConn = HBaseConn.getHBaseConn();Admin admin = hBaseConn.getAdmin()) {TableName name = table.getName();HTableDescriptor descriptor = admin.getTableDescriptor(name);String coprocessorClass = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";if (!descriptor.hasCoprocessor(coprocessorClass)) {admin.disableTable(name);descriptor.addCoprocessor(coprocessorClass);admin.modifyTable(name, descriptor);admin.enableTable(name);}//计时StopWatch stopWatch = new StopWatch();stopWatch.start();Scan scan = new Scan();AggregationClient aggregationClient = new AggregationClient(HBaseConn.getConfiguration());Long count = aggregationClient.rowCount(name, new LongColumnInterpreter(), scan);stopWatch.stop();System.out.println("RowCount：" + count + "，全表count统计耗时：" + stopWatch.getTotalTimeMillis());return count;} catch (Throwable ioe) {ioe.printStackTrace();}return 0L;}
}

HBaseController 测试HBase 的controller

@RestController
@RequestMapping("/hbase")
public class HBaseController {@Resourceprivate RedisUtils redisUtils;/*** 插入轨迹数据* @return return*/@GetMapping("/saveShipPoint")public String saveShipPoint() {// 从redis中取出船舶轨迹点数据List<Object> list = redisUtils.getMaster().getValuesFromPattern("SHIP1:*");if (list == null) {return null;}List<Put> putList = new ArrayList<>(list.size()*2);list.forEach(e->{// 循环遍历JSONObject data = JSONObject.parseObject(e.toString());String shipNumber = data.getString("shipNumber");String shipName = data.getString("shipName");String longitude = data.getString("longitude");String latitude = data.getString("latitude");if(!StringUtils.isEmpty(shipNumber) && !StringUtils.isEmpty(shipName)&& longitude!=null && latitude!=null){// 行键 船舶唯一标识9位码+当前时间Put put = new Put(Bytes.toBytes(shipNumber+"-"+System.currentTimeMillis()));// 船舶九位码put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("shipNumber"), Bytes.toBytes(shipNumber));// 船舶名称（中文）put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("shipName"), Bytes.toBytes(shipName));// 经度put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("longitude"), Bytes.toBytes(longitude));// 纬度put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("latitude"), Bytes.toBytes(latitude));//put插入数据putList.add(put);}});if (!CollectionUtils.isEmpty(putList)) {// 插入ship表中HBaseUtil.putRows("ship", putList);}return  "OK";}/*** 获取ship表中的记录总数* @return ship表中的记录总数*/@GetMapping("/count")public Long getShipCount() {return HBaseUtil.countRowsWithCoprocessor("ship");}/*** 根据时间段查询船舶轨迹* @return re*/@GetMapping("/find")public String find() {//1605843004000 1605843184000 1605843124000// 使用行键前缀过滤器Filter filter = new PrefixFilter(Bytes.toBytes("413790328-"));FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, Collections.singletonList(filter));// 添加时间段条件ResultScanner scanner = HBaseUtil.getScanner("ship", "413790328-1605841204000", "413790328-1605856498271", filterList);AtomicInteger i = new AtomicInteger();// 获取到对应的船舶轨迹listif (scanner != null) {scanner.forEach(result -> {i.getAndIncrement();System.out.println("rowkey=" + Bytes.toString(result.getRow()));System.out.println("fileName=" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("shipName"))));});System.out.println("一共获取数据："+ i.get());// 关闭scanner连接scanner.close();}return "OK"+i.get();}/*** 创建表* @return ship表*/@GetMapping("/createShipTable")public Boolean createShipTable() {return HBaseUtil.createTable("ship", new String[]{"info"});}
}

ShipSchedule 定时器

@Component
@Slf4j
public class ShipSchedule {@ResourceHBaseController hBaseController;// 定时存入轨迹数据//@Scheduled(cron = "0/10 * * * * ? ")public void saveShipPoint(){hBaseController.saveShipPoint();}
}

redis中的测试数据（2万6）

实现步骤

1、访问createShipTable方法创建ship表

2、启动定时器运行saveShipPoint方法，保存船舶得轨迹测试数据（数量达到1000W，大概持续运行一个小时）

3、访问getShipCount方法获取表的总记录数（全表计算用时比较长）

4、访问find方法根据时间段查询船舶轨迹（查询速度达到毫秒级70ms）

5、轨迹数据表可以按天创建每天创建一个ship+日期表（如ship20201122、ship20201123等）。