Spark+TDengine 在中国电信电力测功系统监控平台上的应用实践

小 T 导读:电力测功系统监控平台,是基于中国电信上海理想信息产业(集团)有限公司的,设备数据采集和设备银行V2.0应用进行设计和开发的。采集到的设备的实时数据,都是存储在TDengine Database中的。

应用背景

电力测功系统监控平台,是基于中国电信上海理想信息产业(集团)有限公司的,设备数据采集和设备银行V2.0应用,进行开发和设计的,软件是电力测功系统监控平台的核心,是实现数据汇聚、数据处理、可视化展示以及测功试验的重要部分,根据电力测功系统集控界面功能需求,可将整个电力测功系统监控平台架构分为四部分:设备层、网络层、平台层以及展示层。整体架构如下图:

TDengine 时序数据库 - Spark+TDengine 在中国电信电力测功系统监控平台上的应用实践 1 3

实时数据是存储在TDengine数据库之中的,和设备银行之间进行通信,数据流程如下:

TDengine 时序数据库 - Spark+TDengine 在中国电信电力测功系统监控平台上的应用实践 image 2

设备银行主要负责模版、设备和报警信息的管理,其可视化界面如下:

TDengine 时序数据库 - Spark+TDengine 在中国电信电力测功系统监控平台上的应用实践 3 3

Spark+TDengine使用过程

1. TDengine的安装  

请参考官方文档:https://tdengine.com/docs/cn/v2.0/

2. 在TDengine中建立测试库和测试表

taos> create database test;
taos>use test;
#这里我们创建一个和tdengine自带库log中log表结构一致的表,后提直接从log.log读数据存储到test.log_cp
taos> create table log_cp(
   ->  ts TIMESTAMP,
   ->  level TINYINT,
   ->  content BINARY(80),
   ->  ipaddr BINARY(15)
   -> )

3. Spark读取TDengine  

TDengine虽未提供Spark调用的DataSource,但TDengine本身也支持JDBC,因此,这里使用spark-jdbc来读取TDengine,最新版本可以到官网下载,我这里用的是如下版本: 

<dependency>
        <groupId>com.taosdata.jdbc</groupId>
        <artifactId>taos-jdbcdriver</artifactId>
        <version>1.0.3</version>
    </dependency>

关于JDBC的使用,官网有如下提示: 

由于TDengine是使用C语言开发的,使用taos-jdbcdriver驱动包时需要依赖系统对应的本地函数库。 
1. libtaos.so在Linux系统中成功安装TDengine后,依赖的本地函数库libtaos.so文件会被自动拷贝至/usr/lib/libtaos.so,该目录包含在Linux自动扫描路径上,无需单独指定。  
2.taos.dll在windows系统中安装完客户端之后,驱动包依赖的taos.dll文件会自动拷贝到系统默认搜索路径C:/Windows/System32下,同样无需要单独指定。 

第一次使用时,为了保证机器上有libtaos.so或taos.dll,需要在本地安装TDengine客户端(客户端请至TDengine官网下载) ,Spark的读取代码如下:

    val jdbccdf = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:TAOS://192.168.1.151:6030/log")
      .option("driver", "com.taosdata.jdbc.TSDBDriver")
      .option("dbtable", "log")
      .option("user", "root")
      .option("password", "taosdata")
      .option("fetchsize", "1000")
      .load()

4. Spark存TDengine

因为在读TDengine的时候,第一个字段ts会被转换为decimal,但是存储时直接存decimal tdengine是不认的,所以需要将ts进行类型转换

jdbccdf.select(($"ts" / 1000000).cast(TimestampType).as("ts"), $"level", $"content", $"ipaddr")
      .write.format("jdbc")
      .option("url", "jdbc:TAOS://192.168.1.151:6030/test?charset=UTF-8&locale=en_US.UTF-8")
      .option("driver", "com.taosdata.jdbc.TSDBDriver")
      .option("dbtable", "log2")
      .option("user", "root")
      .option("password", "taosdata")
      .mode(SaveMode.Append)
      .save()

5. Spark yarn模式运行TDengine

上面的测试都是基于maser为local测试的,如果以yarn模式运行,则在每个节点上都安装TDengine客户端是不现实的,查看taos-jdbcdriver的代码,发现,driver会执行System.load(“taos”),也就是说只要java.library.path中存在 libtaos.so,程序就可正常运行,不必安装TDengine的客户端,因为java.library.path是在jvm启动时就设置好的,要更改它的值,可以采用动态加载,采用如下方法解决了加载libtaos.so的问题:

(1) 将driver端libtaos.so发送到各个executor

spark.sparkContext.addFile("/path/to/libtaos.so")

(2) 重写Spark中JdbcUtils类中的createConnectionFactory方法,添加

loadLibrary(new File(SparkFiles.get("libtaos.so")).getParent)

进行java.library.path的动态加载

def createConnectionFactory(options: JDBCOptions): () => Connection = {
    val driverClass: String = options.driverClass
    () => {
      loadLibrary(new File(SparkFiles.get("libtaos.so")).getParent)
      DriverRegistry.register(driverClass)
      val driver: Driver = DriverManager.getDrivers.asScala.collectFirst {
        case d: DriverWrapper if d.wrapped.getClass.getCanonicalName == driverClass => d
        case d if d.getClass.getCanonicalName == driverClass => d
      }.getOrElse {
        throw new IllegalStateException(
          s"Did not find registered driver with class $driverClass")
      }
      driver.connect(options.url, options.asConnectionProperties)
    }
  }

(3) loadLibrary方法如下

 def loadLibrary(libPath: String): Unit = {
    var lib = System.getProperty("java.library.path")
    val dirs = lib.split(":")
    if (!dirs.contains(libPath)) {
      lib = lib + s":${libPath}"
      System.setProperty("java.library.path", lib)
      val fieldSysPath = classOf[ClassLoader].getDeclaredField("sys_paths")
      fieldSysPath.setAccessible(true)
      fieldSysPath.set(null, null)
    }
  }

在yarn模式下一定要给url设置charset和locale,如

charset=UTF-8&locale=en_US.UTF-8

否则container可能会异常退出。

6. libtaos.so其他加载方式  

本来还尝试了jna加载libtaos.so的方式,此方式只需将libtaos.so放入项目resources中,程序变回自动搜索到so文件,奈何不会改TDengine Database中C的代码。

作者简介董鸿飞,大数据开发工程师,2015年加入上海理想大数据实施部,工作至今。目前主要负责公司数据总线产品设计和开发。

公司简介:中国电信上海理想信息产业(集团)有限公司,成立于1999年,注册资本7000万元,是上海市投资规模较大的信息技术企业之一。通过整合公司内各事业部多年大型项目实施的整体实力,公司着力锻造大型信息化项目咨询规划和顶层设计能力,构建“智慧社区”、“智慧园区”及“智慧政务”、“智慧医疗”、“智慧物流”等各类智慧行业应用等整体解决方案,可提供IT外包服务和网络监控运维管理一站式安全解决方案,逐步形成“智慧城市”专业领域产品研发积累和项目交付与平台运营经验,锻造了整体科研队伍和项目实施团队的综合实力。