背景

物联网时序场景是目前最火热的方向之一。海量的时序数据如汽车轨迹数据、汽车状态监控数据、传感器实时监控数据需要存放进入数据库。一般这类场景下存在如下需求

• 数据高写入，低读取
• 需要对写入数据进行基础的图表展示

• 对写入数据进行聚合分析

传统的关系型数据库并不适合此类场景，时序数据库脱颖而出。表格存储时序实例支持时序数据的存储，其具有如下特点：

• Serverless，分布式，低成本
• 高写入支持

• 优秀的索引能力对数据读取数据分析提供了保障。

本文将以车联网为例介绍如何将 Kafka 中的时序数据写入表格存储并进行读取查询。

常见架构

常见的时序 IOT 场景架构如下。

设备如车辆、各类传感器等在物联网平台上进行注册、登录、消息发布等操作。这些平台会基于对应事件、主题对消息进行 ETL 处理并转发。一般情况下消息会被写入如Kafak的消息队列等待消费。如果有流式数据处理计算的需求，那么会由Flink来消费 Kafka 中的数据。而如果需要直接将这些数据储存，则可以直接利用 Kafka 时序的 sink connector 将数据写入 Tablestore 中的时序表中。而我们可以利用 Tablestore 自身提供的 SQL 能力、索引能力，对这部分时序数据进行展示、分析。

Kafka-connect-tablestore说明

Kafka-connect-tablestore 是阿里云 Tablestore 团队开源出的 Kafka sink connector 组件。它包含 Kafka 官方包中 SinkConnector 接口的的一个实现，支持将 Kafka 数据导入表格存储。该组件既支持写入 Tablestore 普通表也支持通过配置，可以将 Json 数据形式的时序数据导入表格存储中的时序表中。Json 中的字段可以通过配置映射成为表格存储时序表中的字段。

可以根据配置将其各字段映射到时序表中的字段中，给出一个示例 Json，映射如下。

{
    "m": "vehicle",               // 通过配置映射为measurement，表示一个度量类别，一个时序表可以存储多种度量类别的数据。
    "d": "vehicle01",             // 通过配置映射为data source，表示一个数据源的id，这里使用车辆id
    "region": "shanghai",         // 通过配置映射为tags中的一个标签，标签的key为region
    "timestamp": 1638868699090,   // 映射为_time字段，记录的时间
    "speed": 55,                  // 通过配置映射为一个field
    "temperature": "20"           // 通过配置映射为一个field
}

开源项目地址见github地址。

下面将以车联网场景为例，演示下如果使用 Kafka + Tablestore 完成时序数据的入库和分析展示工作。

车联网场景测试

场景说明

车联网是 IOT 领域下的一个典型场景，车辆在行程中实时上报位置、里程等信息。下面我们将模拟车辆向 Kafka 实时上传时序数据，然后由 Kakfa 的 Tablestore connector 将数据写入时序表，最后在 Tablestore 中对数据进行查看和分析。

数据结构

车联网可以关注的参数有很多，位置、温度、油量、速度等等。这里测试我们选取如下几个参数温度、地理位置、总里程数、实时速度进行上传。_data_source用来记录设备标识，在车联网场景中，使用车辆 id 填入 _data_source 字段。_tag 字段中记录行程 id，这样可以区分同一辆汽车的不同行程。

参数字段	说明	Tablestore 时序表中对应数据
measurement	记录类型，本例中使用" vehicle"作为这个参数的值	_m_name
vehicle	车辆id	_data_source
tripId	行程id	_tag中的tripId字段
timestamp	当前时间戳	_time
temperature	车内温度	field中的数据
location	地理位置经纬度，格式为"x,y"	field中的数据
miles	总里程	field中的数据
speed	速度	field中的数据

参数配置

按照表格存储官网上 Kafka 时序 Connector的使用说明进行 Kafka 部署、表格存储开通、配置等工作。

基于以上数据结构，我们对 Kafka Connector 配置如下：

# 设置连接器名称。
name=tablestore-sink
# 指定连接器类。
connector.class=TableStoreSinkConnector

# 指定导出数据的Kafka的Topic列表。
topics=test
# 以下为Tablestore连接参数的配置。
# Tablestore实例的Endpoint。 
tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
# 填写AccessKey ID和AccessKey Secret。
tablestore.access.key.id=xxx
tablestore.access.key.secret=xxx
# Tablestore实例名称。
tablestore.instance.name=xxx

table.name.format=<topic>

# 是否自动创建目标表，默认值为false。
auto.create=true

runtime.error.tolerance=all
runtime.error.mode=ignore

# connector工作模式,默认为normal
tablestore.mode=timeseries
# 时序表主键字段映射
tablestore.timeseries.test.measurement=measurement
tablestore.timeseries.test.dataSource=vehicle
tablestore.timeseries.test.tags=tripId
# 时序表时间字段映射
tablestore.timeseries.test.time=timestamp
tablestore.timeseries.test.time.unit=MILLISECONDS
#field字段类型配置
tablestore.timeseries.test.field.name=temperature,location,miles,speed
tablestore.timeseries.test.field.type=double,string,double,double

写入程序

使用 Java 程序模拟向 Kafka 中写入 Json 数据，进而写入 Tablestore。程序启动 100 个任务，每个任务每秒钟生成一条记录进行上报，模拟 100 辆汽车每秒上报一条时序数据。

连接 Kafak 代码如下：

public void init() {

        Properties properties = new Properties();
        //broker的地址清单，建议至少填写两个，避免宕机
        properties.put("bootstrap.servers", "#########:9092");
        properties.put("acks", "all");
        properties.put("retries", 3);
        properties.put("batch.size", 16);
        properties.put("linger.ms", 0);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(properties);
    }

每秒上报数据代码：

   public void upload() {

       int num = 100;
       service = Executors.newScheduledThreadPool(num);

       final Map<String, AutomobileBean> preBeanMap = new ConcurrentHashMap<>();
       final Map<String, NextMove> moveMap = new ConcurrentHashMap<>();

       for (int i = 0; i < num; i++) {
           final int j = i;
           double oilConsumption =  r.nextDouble() * 0.07 + 0.05; // 油耗
           int speed = r.nextInt(80) + 40;
           service.scheduleAtFixedRate(()->{
               try {
                   VehicleRecord msg = getVehicleRecord("vehicle" + j, "" + j, speed, oilConsumption, preBeanMap, moveMap);
                   String jsonStr = map.writeValueAsString(msg);

                   Future<RecordMetadata> future = producer.send(new ProducerRecord<>("test", jsonStr));
                   future.get();
                   System.out.println("Sent:" + jsonStr);
               } catch (Exception e) {
                   e.printStackTrace();
               }
           },1,1, TimeUnit.SECONDS);
       }
   }

控制台查询

表数据查询

可以直接通过控制台中的界面读取写入的时序数据。测试中，模拟了 100 辆汽车上报数据。在控制台选择一条时间线，点击查询数据，进一步点击查询，可以看到写入的数据以列表的形式展示。

SQL 查询

表格存储具备强大的 SQL 能力，可以支持用户使用 SQL 直接筛选、分析数据。

执行以下 SQL，拿到 tripId 为1且记录项为总里程的数据，按时间倒序排列。

select count(*) from test where  tag_value_at(_tags, 'tripId')="1"  
and _field_name = "miles"
order by _time desc limit 1000

可以在控制台的 SQL 页面看到执行结果如下。

执行以下SQL，统计在行程 id 为1的行驶过程中，共上报多少个位置记录。

select count(*) from test where  tag_value_at(_tags, 'tripId')="1"  
and _field_name = "location"

可以在控制台看到结果。

执行以下SQL，可以通过元数据表查看上报数据的车辆列表。

select distinct _data_source from `test::meta` limit 1000

可以看到如下结果

Grafana查询

表格存储同样支持通过 Grafana 查询时序表中的数据。

经过配置，我们可以在 Grafana 面板中选择需要查询的车辆，然后就可以看到其各个时序维度上如总里程、油量、温度、时速，数值随时间变化的曲线图。

表格存储的 SQL 支持时序元数据检索，我们利用这一能力在 Grafana 中增加一个车辆 ID 的变量，如下图左上角名称为 _data_source 的变量。切换这一变量，选择不同车辆 ID，可以看到不同车辆的数据。

总结

本文介绍了在物联网场景下使用 Tablestore 时序表存储、分析时序数据的方案。使用 Kafka 将海量时序数据导入 Tablestore 时序表中，利用 Tablestore Serverless、分布式、高写入等特点对数据进行存储，利用其多元索引能力、SQL 能力对数据进行展示、分析。文中以物联网中典型场景车联网为例，模拟了在这种架构下，车辆上报数据并对上报数据进行分析的过程。

希望本次分享对你的时序设计架构有所帮助，如果希望继续交流，可以加入我们的开发者技术交流群，可搜索群号『11789671』或『23307953』，亦可直接扫码加入。