小 T 导读：电力测功系统监控平台，是基于中国电信上海理想信息产业（集团）有限公司的，设备数据采集和设备银行V2.0应用进行设计和开发的。采集到的设备的实时数据，都是存储在TDengine Database中的。
应用背景
电力测功系统监控平台，是基于中国电信上海理想信息产业（集团）有限公司的，设备数据采集和设备银行V2.0应用，进行开发和设计的，软件是电力测功系统监控平台的核心，是实现数据汇聚、数据处理、可视化展示以及测功试验的重要部分，根据电力测功系统集控界面功能需求，可将整个电力测功系统监控平台架构分为四部分：设备层、网络层、平台层以及展示层。整体架构如下图：
实时数据是存储在TDengine数据库之中的，和设备银行之间进行通信，数据流程如下：
设备银行主要负责模版、设备和报警信息的管理，其可视化界面如下：
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外包服务和网络监控运维管理一站式安全解决方案，逐步形成“智慧城市”专业领域产品研发积累和项目交付与平台运营经验，锻造了整体科研队伍和项目实施团队的综合实力。
