小 T 導(dǎo)讀：電力測功系統(tǒng)監(jiān)控平臺(tái)，是基于中國電信上海理想信息產(chǎn)業(yè)（集團(tuán)）有限公司的，設(shè)備數(shù)據(jù)采集和設(shè)備銀行V2.0應(yīng)用進(jìn)行設(shè)計(jì)和開發(fā)的。采集到的設(shè)備的實(shí)時(shí)數(shù)據(jù)，都是存儲(chǔ)在TDengine Database中的。

應(yīng)用背景

電力測功系統(tǒng)監(jiān)控平臺(tái)，是基于中國電信上海理想信息產(chǎn)業(yè)（集團(tuán)）有限公司的，設(shè)備數(shù)據(jù)采集和設(shè)備銀行V2.0應(yīng)用，進(jìn)行開發(fā)和設(shè)計(jì)的，軟件是電力測功系統(tǒng)監(jiān)控平臺(tái)的核心，是實(shí)現(xiàn)數(shù)據(jù)匯聚、數(shù)據(jù)處理、可視化展示以及測功試驗(yàn)的重要部分，根據(jù)電力測功系統(tǒng)集控界面功能需求，可將整個(gè)電力測功系統(tǒng)監(jiān)控平臺(tái)架構(gòu)分為四部分：設(shè)備層、網(wǎng)絡(luò)層、平臺(tái)層以及展示層。整體架構(gòu)如下圖：

實(shí)時(shí)數(shù)據(jù)是存儲(chǔ)在TDengine數(shù)據(jù)庫之中的，和設(shè)備銀行之間進(jìn)行通信，數(shù)據(jù)流程如下：

設(shè)備銀行主要負(fù)責(zé)模版、設(shè)備和報(bào)警信息的管理，其可視化界面如下：

Spark+TDengine使用過程

1. TDengine的安裝  

請(qǐng)參考官方文檔：https://tdengine.com/docs/cn/v2.0/

2. 在TDengine中建立測試庫和測試表

taos> create database test;
taos>use test;
#這里我們創(chuàng)建一個(gè)和tdengine自帶庫log中l(wèi)og表結(jié)構(gòu)一致的表，后提直接從log.log讀數(shù)據(jù)存儲(chǔ)到test.log_cp
taos> create table log_cp(
   ->  ts TIMESTAMP,
   ->  level TINYINT,
   ->  content BINARY(80),
   ->  ipaddr BINARY(15)
   -> )

3. Spark讀取TDengine  

TDengine雖未提供Spark調(diào)用的DataSource，但TDengine本身也支持JDBC，因此，這里使用spark-jdbc來讀取TDengine，最新版本可以到官網(wǎng)下載，我這里用的是如下版本： 

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

關(guān)于JDBC的使用，官網(wǎng)有如下提示： 

由于TDengine是使用C語言開發(fā)的，使用taos-jdbcdriver驅(qū)動(dòng)包時(shí)需要依賴系統(tǒng)對(duì)應(yīng)的本地函數(shù)庫。 
1. libtaos.so在Linux系統(tǒng)中成功安裝TDengine后，依賴的本地函數(shù)庫libtaos.so文件會(huì)被自動(dòng)拷貝至/usr/lib/libtaos.so，該目錄包含在Linux自動(dòng)掃描路徑上，無需單獨(dú)指定。  
2.taos.dll在windows系統(tǒng)中安裝完客戶端之后，驅(qū)動(dòng)包依賴的taos.dll文件會(huì)自動(dòng)拷貝到系統(tǒng)默認(rèn)搜索路徑C:/Windows/System32下，同樣無需要單獨(dú)指定。 

第一次使用時(shí)，為了保證機(jī)器上有l(wèi)ibtaos.so或taos.dll，需要在本地安裝TDengine客戶端(客戶端請(qǐng)至TDengine官網(wǎng)下載) ，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

因?yàn)樵谧xTDengine的時(shí)候，第一個(gè)字段ts會(huì)被轉(zhuǎn)換為decimal，但是存儲(chǔ)時(shí)直接存decimal tdengine是不認(rèn)的，所以需要將ts進(jìn)行類型轉(zhuǎn)換

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模式運(yùn)行TDengine

上面的測試都是基于maser為local測試的，如果以yarn模式運(yùn)行，則在每個(gè)節(jié)點(diǎn)上都安裝TDengine客戶端是不現(xiàn)實(shí)的，查看taos-jdbcdriver的代碼，發(fā)現(xiàn)，driver會(huì)執(zhí)行System.load(“taos”)，也就是說只要java.library.path中存在 libtaos.so，程序就可正常運(yùn)行，不必安裝TDengine的客戶端，因?yàn)閖ava.library.path是在jvm啟動(dòng)時(shí)就設(shè)置好的，要更改它的值，可以采用動(dòng)態(tài)加載，采用如下方法解決了加載libtaos.so的問題：

(1) 將driver端libtaos.so發(fā)送到各個(gè)executor

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

(2) 重寫Spark中JdbcUtils類中的createConnectionFactory方法，添加

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

進(jìn)行java.library.path的動(dòng)態(tài)加載

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設(shè)置charset和locale，如

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

否則container可能會(huì)異常退出。

6. libtaos.so其他加載方式  

本來還嘗試了jna加載libtaos.so的方式，此方式只需將libtaos.so放入項(xiàng)目resources中，程序變回自動(dòng)搜索到so文件，奈何不會(huì)改TDengine Database中C的代碼。

作者簡介：董鴻飛，大數(shù)據(jù)開發(fā)工程師，2015年加入上海理想大數(shù)據(jù)實(shí)施部，工作至今。目前主要負(fù)責(zé)公司數(shù)據(jù)總線產(chǎn)品設(shè)計(jì)和開發(fā)。

公司簡介：中國電信上海理想信息產(chǎn)業(yè)（集團(tuán)）有限公司，成立于1999年，注冊(cè)資本7000萬元，是上海市投資規(guī)模較大的信息技術(shù)企業(yè)之一。通過整合公司內(nèi)各事業(yè)部多年大型項(xiàng)目實(shí)施的整體實(shí)力，公司著力鍛造大型信息化項(xiàng)目咨詢規(guī)劃和頂層設(shè)計(jì)能力，構(gòu)建“智慧社區(qū)”、“智慧園區(qū)”及“智慧政務(wù)”、“智慧醫(yī)療”、“智慧物流”等各類智慧行業(yè)應(yīng)用等整體解決方案，可提供IT外包服務(wù)和網(wǎng)絡(luò)監(jiān)控運(yùn)維管理一站式安全解決方案，逐步形成“智慧城市”專業(yè)領(lǐng)域產(chǎn)品研發(fā)積累和項(xiàng)目交付與平臺(tái)運(yùn)營經(jīng)驗(yàn)，鍛造了整體科研隊(duì)伍和項(xiàng)目實(shí)施團(tuán)隊(duì)的綜合實(shí)力。