背景
物联网时序场景是目前最火热的方向之一。海量的时序数据如汽车轨迹数据、汽车状态监控数据、传感器实时监控数据需要存放进入数据库。一般这类场景下存在如下需求
-
数据高写入，低读取
-
需要对写入数据进行基础的图表展示
-
对写入数据进行聚合分析
传统的关系型数据库并不适合此类场景，时序数据库脱颖而出。表格存储时序实例支持时序数据的存储，其具有如下特点：
-
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，这样可以区分同一辆汽车的不同行程。
参数配置
按照表格存储官网上 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』，亦可直接扫码加入。
