共计 5526 个字符，预计需要花费 14 分钟才能阅读完成。
【作者介绍】：大罗，黑格智造架构师，次要从事云原生，大数据系统开发，曾参加国家示范级工业互联网零碎建设等。
做工业互联网或物联网零碎，最根本的需要是展现数据曲线，比方功率曲线，相似于股票的分时图，通常咱们会取每分钟内该设施上报的最初一次功率值为这一分钟的功率，如果某一分钟内，设施没有上报，则取上一分钟的功率值，以此类推。举例如下：
失去的分钟曲线：
通常咱们会把设施上报的数据先写入 Apache Kafka。如果是离线计算场景，可能会思考把数据写入 Hive，而后应用 Spark SQL 定时读取 Hive，再把计算结果写入 HBase；如果是实时计算场景，则会应用 Apache Flink 生产 Kafka 数据，把后果写入 HBase，这种状况下还须要思考数据乱序和提早投递计算等问题。
而且，基于传统大数据 Hadoop 的架构，须要搭建 ZooKeeper 和 HDFS，而后才是 Hive 和 HBase，整个体系保护老本很高。此外，HBase 基于键值存储时序数据，会节约很多空间在同一键值的数据设计架构下面。
以上所举，是物联网设施属性曲线计算场景的其中一个痛点，另外还须要思考数据增长、数据核查以及数据容灾等特点。
笔者所在的公司，要基于 3D 打印技术给客户提供整体化解决方案，天然须要对设施的运行状态做继续追踪，须要存储设备的运行数据。这时候咱们找到了开源的物联网大数据平台 TDengine（https://github.com/taosdata/TDengine）。
参考 TDengine 的文档中 SQL 的写法，在数据齐全的状况下，能够轻松地用一句 SQL 解决下面的问题：
select last(val) a from super_table_xx where ts >= '2021-06-07 18:10:00' and ts <= '2021-06-07 18:20:00' interval(60s) fill(value, 0);
为什么相似的 SQL，TDengine 的执行效率能够如此之高呢？
这就在于它的超级表以及子表，针对单个设施的数据，TDengine 设计了依照工夫间断存储的个性。而事实上，业务零碎在应用物联网数据的时候，无论是即时查问还是离线剖析，存在读取单个设施的一个间断时间段数据的特点。
假如，咱们要存储设备的温度与湿度，咱们能够设计超级表如下：
create stable if not exists s_device (ts TIMESTAMP,
temperature double,
humidity double
) TAGS (device_sn BINARY(1000));理论应用中，例如针对设施’d1’和’d2’的数据执行插入的 SQL 如下：
insert into s_device_d1 (ts, temperature, humidity) USING s_device (device_sn) TAGS ('d1') values (1623157875000, 35.34, 80.24);
insert into s_device_d2 (ts, temperature, humidity) USING s_device (device_sn) TAGS ('d2') values (1623157891000, 29.63, 79.48);搜寻设施’d1’某个时间段的数据，其 SQL 如下：
select * from s_device where device_sn = 'd1' and ts > 1623157871000 and ts < 1623157890000 ;假如统计过来 7 天的平均温度曲线，每小时 1 个点:
select avg(temperature) temperature from s_device where device_sn = #{deviceSn} and ts >= #{startTime} and ts < #{endTime} interval(1h)
TDengine 还提供了很多聚合函数，相似下面的计算 1 分钟间断曲线的 last 和 fill，以及其余罕用的 sum 和 max 等。
在和应用程序联合的过程中，咱们选用 MyBatis 这种灵便易上手的 ORM 框架，例如，针对下面的数据表’s_device’，咱们先定义 entity :
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.sql.Timestamp;
/**
* @author: DaLuo
* @date: 2021/06/25
* @description:
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
@TableName(value = "s_device")
public class TestSuperDeviceEntity {
private Timestamp ts;
private Float temperature;
private Float humidity;
@TableField(value = "device_sn")
private String device_sn ;
}再定义 mapper：
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.hg.device.kafka.tdengine.entity.TestSuperDeviceEntity;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import java.sql.Timestamp;
import java.util.List;
/**
* @author: DaLuo
* @date: 2021/06/25
* @description:
*/
@Mapper
public interface TestSuperDeviceMapper extends BaseMapper<TestSuperDeviceEntity> {
/**
* 单个插入
* @param entity
* @return
*/
@Insert({"INSERT INTO's_device_${entity.deviceSn}'(ts ,temperature, humidity)",
"USING s_device (device_sn) TAGS (#{entity.deviceSn})",
"VALUES (#{entity.ts}, #{entity.temperature}, #{entity.humidity})"
})
int insertOne(@Param(value = "entity") TestSuperDeviceEntity entity);
/**
* 批量插入
* @param entities
* @return
*/
@Insert({
"<script>",
"INSERT INTO",
"<foreach collection='list'item='item'separator=' '>",
"'s_device_${item.deviceSn}' (ts ,temperature, humidity) USING s_device (device_sn) TAGS (#{item.deviceSn}) ","VALUES (#{item.ts}, #{item.temperature}, #{item.humidity})","</foreach>","</script>"
})
int batchInsert(@Param("list") List<TestSuperDeviceEntity> entities);
/**
* 查问过来一段时间范畴的平均温度，每小时 1 个数据点
* @param deviceSn
* @param startTime inclusive
* @param endTime exclusive
* @return
*/
@Select("select avg(temperature) temperature from s_device where device_sn = #{deviceSn} and ts >= #{startTime} and ts < #{endTime} interval(1h)")
List<TempSevenDaysTemperature> selectSevenDaysTemperature(@Param(value = "deviceSn") String deviceSn,
@Param(value = "startTime") long startTime,
@Param(value = "endTime") long endTime);
@AllArgsConstructor
@NoArgsConstructor
@Data
@Builder
class TempSevenDaysTemperature {
private Timestamp ts;
private float temperature;
}
}TDengine 有一个很奇妙的设计，就是不必事后创立子表，所以咱们能够很不便地利用’tag’标签作为子表名称的一部分，即时插入数据同时创立子表。
留神：思考到跨时区的国际化个性，咱们所有的工夫存储查问交互，都是应用的工夫戳，而非”yyyy-mm-dd hh:MM:ss”格局，因为数据存储波及到应用程序时区，连贯字符串时区，TDengine 服务时区，应用”yyyy-mm-dd hh:MM:ss”格局容易导致工夫存储的不准确性，而应用工夫戳，长整型的数据格式则能够完满地防止此类问题。
Java 应用 TDengine JDBC-driver 目前有两种形式：JDBC-JNI 和 JDBC-RESTful，前者在写入性能上更有劣势。然而须要在利用程序运行的服务器上安装 TDengine 客户端驱动。
咱们的应用程序用到了 Kubernetes 集群，程序是运行在 Docker 外面，为此咱们制作了一个适宜咱们利用程序运行的镜像，例如根底镜像的 Dockerfile 如下所示：
FROM openjdk:8-jdk-oraclelinux7
COPY TDengine-client-2.0.16.0-Linux-x64.tar.gz /
RUN tar -xzvf /TDengine-client-2.0.16.0-Linux-x64.tar.gz && cd /TDengine-client-2.0.16.0 && pwd && ls && ./install_client.shbuild：
docker build -t tdengine-openjdk-8-runtime:2.0.16.0 -f Dockerfile .
援用程序镜像 Dockerfile 所示：
FROM tdengine-openjdk-8-runtime:2.0.16.0
ENV JAVA_OPTS="-Duser.timezone=Asia/Shanghai -Djava.security.egd=file:/dev/./urandom"
COPY app.jar /app.jar
ENTRYPOINT ["java","-jar","/app.jar"]这样咱们的应用程序就能够调度在任意的 K8s 节点上了。
另外，咱们的程序波及到工作自动化调度，须要频繁地和设施下位机进行 MQTT 数据交互，比方，云端发送指令 1000-“开始工作 A”，下位机回复指令 2000-“收到工作 A”，把指令了解成设施，把指令序列号以及内容了解成它的属性，天然这种数据也是非常适合存储在 TDengine 时序数据库中的：
*************************** 1.row ***************************
ts: 2021-06-23 16:10:30.000
msg: {"task_id":"7b40ed4edc1149f1837179c77d8c3c1f","action":"start"}
device_sn: deviceA
kind: 1000
*************************** 2.row ***************************
ts: 2021-06-23 16:10:31.000
msg: {"task_id":"7b40ed4edc1149f1837179c77d8c3c1f","action":"received"}
device_sn: deviceA
kind: 2000咱们云端在和设施对接的过程中，频繁须要讲究音讯是否发送的问题，所以迫切需要对指令进行保留，从而在应用程序中新辟线程，专门订阅指令集音讯，批量写入到 TDengine 数据库。
最初，TDengine 还有一个超级表 log.dn，外面保留了内存、CPU 等应用信息，所以咱们能够利用 Grafana 展现这些数据，为监控提供牢靠的经营数据参照！