Spark 读取、写入时序数据库TDengine以及TDengine概述

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

一、TDengine是什么

TDengine 是一款高性能、分布式、支持 SQL 的时序数据库其核心代码包括集群功能全部开源开源协议AGPL v3.0。TDengine 能被广泛运用于物联网、工业互联网、车联网、IT 运维、金融等领域。除核心的时序数据库功能外TDengine 还提供缓存、数据订阅、流式计算等大数据平台所需要的系列功能最大程度减少研发和运维的复杂度。

1.TDengine总结出了物联网数据的特点

数据是时序的一定带有时间戳

数据是结构化的

数据极少有更新或删除操作

数据源是唯一的

相对互联网应用写多读少

用户关注的是一段时间的趋势而不是某一特点时间点的值

数据是有保留期限的

数据的查询分析一定是基于时间段和地理区域的

除存储查询外还往往需要各种统计和实时计算操作

流量平稳可以预测

往往需要有插值等一些特殊的计算

数据量巨大一天采集的数据就可以超过100亿条。

总结出这些特点后会发现物联网数据像日志数据几乎没有更新操作的可能那数据库中的事务处理的实现就完全多余数据是时序的时间戳自然可以作为主键根本不需要复杂的索引结构

物联网数据是结构化的像HBase、Cassandra那样用Key-Value来存储计算效率和存储效率大打折扣应该采用结构化存储才行

物联网数据的冷热程度是时间决定的刚采集的数据是最热的而不是用户点击决定。因此用简单的先进先出的内存管理就能很好的实现高效的缓存根本就不需要Redis

物联网数据从一个设备来看就是一个数据流实现滑动窗口的流计算那不是一个最顺其自然的事情哪用得上Spark这么复杂的引擎

对于数据分区简单的按设备分区按时间段分区就轻松解决根本就不需要复杂的分区机制

物联网数据流是相对平稳的而且物联网设备本身一定有缓存能力完全可以抛弃Kafka这些套件实现一个简单的消息队列和数据订阅来满足需求。

随后又发现有时序数据库马上看他们文档和代码发现他们利用了时序数据一些特点但还是没有充分利用而且只是定位为一个数据库。

后面又了解到工业界有实时数据库发现这些实时数据库都是老古董产品了基本上都还是Windows上研发价格贵而且没有标准SQL水平扩展几乎没有大数据分析能力几乎没有完全没有能力应对日益增长的大数据量和大数据分析需求迟早会被淘汰。

一个分布式高可靠、持久化存储的消息队列每台手机需要推送的消息就放在一个队列里。消息队列与物联网的时序数据有区别吗本质上没有。

一个是非结构化的一个是结构化的

一个是简单的进和出但另外一个是需要有分析和计算的

两者在系统的架构设计上没有大的不同。

因此TDengine迅速定位了自己要做的产品那就是物联网大数据平台要把时序数据库、缓存、消息订阅、流式计算等系列功能融合在一起一站式的解决物联网大数据问题这样才能将系统研发、维护的复杂度与成本大幅降低。

研究完物联网数据的特点后TDengine做了两个技术创新点

2.TDengine的开源协议

TDengine目前已经在GitHub上进行了开源。采用是AGPL许可。AGPL的全称为GNU Affero General Public License是GPL的一个补充在GPL的基础上加了一些限制。

开源项目地址https://github.com/taosdata/TDengine

3.TDengine的优势

10 倍以上的性能提升定义了创新的数据存储结构单核每秒能处理至少 2 万次请求插入数百万个数据点读出一千万以上数据点比现有通用数据库快十倍以上。

硬件或云服务成本降至 1/5由于超强性能计算资源不到通用大数据方案的 1/5通过列式存储和先进的压缩算法存储占用不到通用数据库的 1/10。

全栈时序数据处理引擎将数据库、消息队列、缓存、流式计算等功能融合应用无需再集成Kafka/Redis/HBase/Spark/HDFS 等软件大幅降低应用开发和维护的复杂度成本。与第三方工具无缝连接不用一行代码即可与 Telegraf, Grafana, Matlab, R 集成。后续将支持 MQTT, OPC, Hadoop,Spark等, BI工具也将无缝连接。

强大的分析功能无论是十年前还是一秒钟前的数据指定时间范围即可查询。数据可在时间轴上或多个设备上进行聚合。即席查询可通过 Shell, Python, R, MATLAB 随时进行。

高可用性和水平扩展通过分布式架构和一致性算法通过多复制和集群特性TDengine确保了高可用性和水平扩展性以支持关键任务应用程序。

零运维成本、零学习成本安装集群简单快捷无需分库分表实时备份。类似标准 SQL支持 RESTful支持 Python/Java/C/C++/C#/Go/Node.js, 与 MySQL 相似零学习成本。

核心开源除了一些辅助功能外TDengine的核心是开源的。企业再也不会被数据库绑定了。这使生态更加强大产品更加稳定开发者社区更加活跃。

4.TDengine的适用场景

TDengine作为一个基础性的软件应用范围及其广泛原则上所有使用机器、设备、传感器采集数据的地方都可以用上。

一些典型场景罗列如下

公共安全上网记录、通话记录、个体追踪、区间筛选

电力行业智能电表、电网、发电设备的集中监测

通讯行业话费详单、用户行为、基站/通讯设备监测

金融行业交易记录、存取记录、ATM、POS 机监测

出行工具火车/汽车/出租/飞机/自行车的实时监测

交通行业实时路况路口流量监测卡口数据

石油石化油井、运输管线、运输车队的实时监测

互联网服务器/应用监测、用户访问日志、广告点击日志

物流行业车辆、集装箱的追踪监测

环境监测天气、空气、水文、地质环境等监测

物联网电梯、锅炉、机械、水表、气表等各种联网设备

军工行业各种军事装备的数据采集、存储

制造业生产过程管控流程数据、供应链数据采集与分析

5.TDengine的特点

专为物联网数据而设计利用物联网数据时序性的特点实现每个采集点对应一个表的这一功能。但其不适用于处理通用的互联网数据。

采用列式存储+压缩的方式以节省硬件成本。(压缩效率高利用物联网数据变化波动不大的特性、dif插值后压缩然后二阶压缩效率很高。)

支持高可用性将每个物理节点划分为多个虚拟数据节点和虚拟管理节点。虚拟数据节点存储数据虚拟管理节点管理MetaData。虚拟数据节点和虚拟管理节点分布在不同的物理节点上实现数据集应用的高可用。

存储结构上采用每个采集点创建一个独立的表的方式来存储。这样实现每个采集点的数据的连续存放提升读取效率。由于每个表的数据来源只有一个能够实现无锁写入提升写入速率。

针对多变的聚合引入了超级表的概念。同一类型的采集设备可以创建一张超级表。在创建超级表时可以对这类表指定标签在查询的时候通过标签来对数据库中的表进行过滤这样即使数据库中有非常多的表也可以实现快速的多表聚合。

安装包非常小安装使用简单。支持SQL语法与MySQL类似。

二、TDengine 安装

安装包地址https://www.taosdata.com/all-downloads

代码案例地址 https://www.cnblogs.com/jason-dong/p/13127459.html

1.安装步骤

准备工作要主机名映射/etc/hosts

解压TDengine-server-2.4.0.10-Linux-x64.tar.gz解压的文件夹下有个install.sh命令直接运行即可

linux默认安装位置/etc/taos/

这里已经安装好了已经可以启动使用了(默认端口6030)安装就是这么简单

systemctl start taosd //启动

systemctl status taosd //查看状态

systemctl stop taosd //停止

#进入taos 默认用户名root 默认密码taosdata

taos -u root -p

CREATE TABLE taos001 (ts timestamp, id int, name NCHAR(255), sex BOOL);

TDengine 在 SQL 层面并没有提供修改与删除操作的语句即不存在 UPDATE , DELETE 语句

TDengine默认配置下数据不可更新

TDengine默认配置下数据存储10年

CREATE DATABASE [IF NOT EXISTS] db_name [KEEP keep] [DAYS days] [UPDATE 1];

KEEP是该数据库的数据保留多长天数缺省是3650天(10年)数据库会自动删除超过时限的数据

UPDATE 标志数据库支持更新相同时间戳数据

建库时不指定 update 参数则 update 默认为0表示数据不可修改相同时间戳的数据直接被丢弃。

可通过 update 参数配置为可修改但是在修改时要求时间戳必须一样其他字段才能修改成功显然时间戳无法修改~。即根据时间戳后来的值把前面的字节覆盖数据还是一条

卸载命令rmtaos

但实际操作过后发现卸载后再重新安装启动还是有问题后来卸载后,使用find / -name 'taos*'找到所有taos文件夹删除后再安装就没问题了(实际应用时绝对不可以使用这种操作)

2.TDengine支持的数据类型

三、Spark操作TDengine

1.pom依赖

<!-- https://mvnrepository.com/artifact/com.taosdata.jdbc/taos-jdbcdriver -->
<dependency>
    <groupId>com.taosdata.jdbc</groupId>
    <artifactId>taos-jdbcdriver</artifactId>
    <version>2.0.36</version>
</dependency>

2.SparkDataFrame读写TDengine

object TDengineRead {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("testTDengineRead")
      .master("local")
      .getOrCreate()

    val driver = "com.taosdata.jdbc.TSDBDriver"
    val userName = "root"
    val password = "taosdata"
    val url = "jdbc:TAOS://192.168.1.1:6030/test"

    val sql = "select * from taos001 "
    val dbtable = "( " + sql + ") temp"
    
    val df1 = spark.createDataFrame(
      Seq(("2023-02-02 11:20:28.125", "5", "ming5", "true"),
        ("2023-02-02 10:20:21", "2", "hong", "true"),
        ("2023-02-02 13:20:22", "3", "zhi", "false")
      )).toDF("ts", "id", "name", "sex")
    df1.show()
    
    println("=========================")

    val properties = new Properties()
    properties.setProperty("user", userName)
    properties.setProperty("password", password)
    properties.setProperty("driver", driver)
    df1.write.mode(SaveMode.Append).jdbc(url, "taos001", properties)
    
    val df2 = spark.read.format("jdbc")
      .option("url", url)
      .option("driver", driver)
      .option("dbtable", dbtable)
      .option("user", userName)
      .option("password", password)
      .option("fetchsize", "500000")
      .load()
    df2.show()    
  }
}