基于TDengine+GeoServer时空大数据云平台

应用简介：

基于 WebGL 技术和 Cesium 开源框架实现，构建数字地球，集成遥感影像、电子海图、陆地/海底地形、倾斜摄影、海洋气象等基础数据，通过接入LD、无线电频谱侦测、光电/红外设备及AIS/ADS-B设备，应用多源异构数据融合技术掌握装备部署及其位置、状态和相互关系，将各设备上报的目标点迹与航迹进行融合形成综合TS时空大数据云平台。运用GIS、大数据技术对多来源数据关联分析，可以准确研判ZC形势，显著增强指挥员的判断力和敏锐性。对ZC环境、目标TS和DC环境等数据的实时融合处理和可视化展示，能够帮助指挥员灵活调整资源分配和BL部署，确保战术应对的灵活性与精准度。基于实时在线大数据分析，指挥员能够精准规划任务、精细计算BL与 HL、对目标移动轨迹预测（基于先验知识的多模态轨迹预测方法）、精确下达ZZ指令，从而动态高效调控队伍行动、准确评估ZZ效果。

应用场景：

综合态势时空云平台处理海量位置关联信息，以开源AIS数据为例，单个季度数据150GB,数据量超过6亿条，这类海量时序数据的存储和计算，有着体量大、时间长，写入，查询要求高的特点，传统关系型数据库无法满足实时写入与高性能查询要求。 TDengine 3.0 引入了GEOMETRY 类型，基于OpenGIS规范，在SQL级别上实现了基本的空间运算功能，与GeoServer结合后，可以利用GeoServer的地图服务功能，实时发布和可视化时空数据；这和时空云平台很契合,例如查询某一条船历史航迹，利用AI预测他的轨迹。例如查看X海热点区域最近3个月空中，水面，水下目标态势回放复盘。

实践方案：

关键核心在于数据存储与管理，考虑后期目标会越来越多（无人机、无人艇增多），数据存储问题 ，高频采集数据入库效率问题，解决海量时序数据的存储和计算，有着体量大、时间长，写入，查询要求高的特点，传统关系型数据库无法满足实时写入与高性能查询要求。我们从主流时序数据库InfluxDB、timescaleDB、TDengine.中选型，从入库效率、压缩比、应用查询场景效率以及团队上手难度四个维度测试对比，TDengine 完胜对手。TDengine展现出了强大的读写性能和数据压缩能力，聚合类查询速度非常快，也帮助我们有效降低了机器使用成本。超级表、子表、标签、时间窗口、状态窗口、GIS空间数据运算等概念非常适配时空大数据应用场景。

创建数据库

以下用南海地区开源离线AIS数据为例创建创建时空数据库，创建航迹点超级表，包含时间点字段精确到ms, 对地航向，速度，高度（深度），网格号，点位置Point,标签存放唯一标识码mmsi, 船类别等其他信息。 超级表中针对全球海量数据，采用分治思想，将全球划分为网格,使用geohash算法生成唯一标识符，利用Geohash的前缀匹配能力，能够快速过滤掉不相关的网格，再使用函数ST_Contains查询范围内的数据，以提高查询效率。

geohash 编码原理图

创建数据库： CREATE DATABASE ship PRECISION 'ms' KEEP 3650 DURATION 30; 创建超级表： create stable trajectory ( ts TIMESTAMP, cog FLOAT,sog FLOAT，height FLOAT ，rc VARCHAR(10) ,geom GEOMETRY(21) ) tags(mmsi VARCHAR(12) ,...);

离线数据入库

开发环境是docker, 开源版本3.3.2.0。 离线数据处理入库使用Spring boot 开发批处理程序，数据库连接方式使用WebSocket方式。

历史数据入库 ,读取离线航迹文件，每读取1000万条记录，执行一次入库操作，采用异步多线程批量高效入库。数据入库前文件143GB，入库后数据目录21.3GB,数据压缩效果1/7很NICE!

数据入库时间接近1小时（程序同时在转AI训练的csv文件会牺牲一点效率），在相对较差的开发环境表现也很亮眼。

   // 每次读取1000万条记录提交任务到线程池异步执行，继续读取下一批次
  if ( index >= 10000000) {   
    indexFile++;
    if(indexFile <= 20) {
         String outputPath="F:\\project\\2024\\AIS\\out_test_"+indexFile+".csv";
         writeMapToCsv(  outputPath, listMap);
         log.info("写完第"+indexFile +"AI航迹预测训练文件文件!!!");
    }
    Map<String, Set<QHAITrajectory.ShipPoint> > map = new HashMap<>();
    map.putAll(listMap);
    executorService.submit(() -> tdHelper.save2Db(map)); // 提交任务到线程池
    index = 0;
    listMap.clear();
 }
 // 批量入库
  try(Connection  connection = getConnection()) {
    for (Map.Entry<String, Set<QHAITrajectory.ShipPoint>> entry : map.entrySet()) {
        String key = entry.getKey();
        String sql = "INSERT INTO ship.s" + key + " USING ship.trajectory TAGS ('" + key + "') VALUES ";
        Set<QHAITrajectory.ShipPoint> shipPoints = entry.getValue();
        //( ts TIMESTAMP, cog FLOAT,sog FLOAT,loc GEOMETRY(21) )
        int index = 0;
        StringBuilder sb = new StringBuilder();
        sb.append(sql);
        for (QHAITrajectory.ShipPoint shipPoint : shipPoints) {
            sb.append("('" + shipPoint.getUpdateDateFormat() + "', " + shipPoint.getCog() + ", " + shipPoint.getSog() + ", " + shipPoint.getGeohash() + ", 'Point(" + shipPoint.getLng() + " " + shipPoint.getLat() + ")' ) ");
            index++;
            // 每写入 1000 条记录后执行一次插入操作
            if (index % BATCH_SIZE == 0) {
                executeBatchQuery(connection, sb.toString());
                sb.setLength(0); // 清空 StringBuilder
                sb.append(sql);
            }
        }
        if (sb.length() > sql.length()) {
            executeBatchQuery( connection,sb.toString());
        }
    }

实时数据入库

数据来源分来源分好几类，AIS数据是Apache RocketMQ数据订阅,XX数据来源HTTP接口同步，XX数据UDP报文接收，还有部分数据TXT文件共享，开发对应数据接收解析程序将数据推送到入库队列，队列内部按照目标分二级队列，入库程序按照目标队列批量入库。

geoserver-2.25-tdengine-plugin插件开发

GeoServer拓展数据源的方式

1 创建一个类实现接口

矢量数据实现：org.geotools.spi.data.DataStoreFactorySpi

2 在目录META-INF/services/下增加文件org.geotools.spi.data.DataStoreFactorySpi，内容为刚创建的类的完全类名org.geotools.data.tdengine.TDengineDataStoreFactory；

  public class TDengineDataStoreFactory extends JDBCDataStoreFactory{
    public static final Param DBTYPE = new Param("dbtype", String.class, "Type", true, "TAOS-RS", Collections.singletonMap("level", "program"));
    public static final Param PORT = new Param("port", Integer.class, "Port", true, 6041);

    @Override
    protected String getDatabaseID() {
        return (String)DBTYPE.sample;
    }
    @Override
    protected String getDriverClassName() {
        return "com.taosdata.jdbc.rs.RestfulDriver";
    }
    @Override
    protected SQLDialect createSQLDialect(JDBCDataStore jdbcDataStore) {
        return new TDengineDialectBasic(jdbcDataStore, true);
    }
    @Override
    protected String getValidationQuery() {
        return "SELECT SERVER_VERSION()";
    }
    @Override
    public String getDescription() {
        return "TDengine Time-Series Database 涛思时序数据库";
    }
    @Override
    protected void setupParameters(Map<String, Object> parameters) {
        super.setupParameters(parameters);
        parameters.put(DBTYPE.key, DBTYPE);
        parameters.put(PORT.key, PORT);
        parameters.remove(SCHEMA.key);
    }
    @Override
    protected String getJDBCUrl(Map<String, ?> params) throws IOException {
        String host = (String)HOST.lookUp(params);
        Integer port = (Integer)PORT.lookUp(params);
        String db = (String)DATABASE.lookUp(params);
        String databaseID = this.getDatabaseID();
        String url = "jdbc:" + databaseID + "://" + host;
        if (port != null) {
            url = url + ":" + port;
        }
        if (db != null) {
            url = url + "/" + db;
        }
        //REST 连接中增加 batchfetch 参数并设置为 true，将开启 WebSocket 连接
        url = url + "?batchfetch=true&charset=UTF-8&timezone=UTC-8&enableAutoReconnect=true";
        return url;
    }
}

3 复制打包的jar文件和依赖的taos-jdbcdriver-3.3.2.jar、fastjson-1.2.83.jar、guava-32.1.3-jre.jar到geoserver\WEB-INF\lib目录下，或者在调试webapp ,模块添加引用。

写在最后

在时空云平台产品中，TDengine展现出了强大的读写性能和数据压缩能力，时间区间查询速度非常快，同时它还带有内建的缓存、流式计算、数据订阅等系统功能，能大幅减少系统设计的复杂度，降低研发和运营成本。TDengine是一个极简的时序数据处理平台，它是我们降本增效利器。特别是TDengine3.0支持了Windows平台，对于我们传统软件厂商GIS行业来说是福音。国家近年来在推国产化软硬件平台，特别是涉密部门只能用国产软件，给了TDengine一个大舞台，展现自己大数据管理能力。

百舸争流，奋楫者先;千帆竟发，勇进者胜；祝涛思数据越来越好，做到时序大数据No.1。