小 T 導讀：想用 Flink 對接 TDengine？保姆級教程來了。

0、前言

TDengine 是由濤思資料開發並開源的一款高效能、分散式、支援 SQL 的時序資料庫（Time-Series Database）。

除了核心的時序資料庫功能外，TDengine 還提供快取、資料訂閱、流式計算等大資料平臺所需要的系列功能。但是很多小夥伴出於架構的考慮，還是需要將資料匯出到 Apache Flink、Apache Spark 等平臺進行計算分析。

為了幫助大家對接，我們特別推出了保姆級課程，包學包會。

1、技術實現

Apache Flink 提供了 SourceFunction 和 SinkFunction，用來提供 Flink 和外部資料來源的連線，其中 SouceFunction 為從資料來源讀取資料，SinkFunction 為將資料寫入資料來源。 與此同時，Flink 提供了 RichSourceFunction 和 RichSinkFunction 這兩個類（繼承自

2、程式碼實現

完整原始碼：https://github.com/liuyq-617/TD-Flink

程式碼邏輯：

1) 自定義類 SourceFromTDengine

用途：資料來源連線，資料讀取

1. package com.taosdata.flink;
2. import org.apache.flink.configuration.Configuration;
3. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
4. import com.taosdata.model.Sensor;
5. import java.sql.*;
6. import java.util.Properties;
8. public class SourceFromTDengine extends RichSourceFunction<Sensor> {
9. Statement statement;
10. private Connection connection;
11. private String property;
12. public SourceFromTDengine(){
13. super();
14. }
16. @Override
17. public void open(Configuration parameters) throws Exception {
18. super.open(parameters);
19. String driver = "com.taosdata.jdbc.rs.RestfulDriver";
20. String host = "u05";
21. String username = "root";
22. String password = "taosdata";
23. String prop = System.getProperty("java.library.path");
24. Logger LOG = LoggerFactory.getLogger(SourceFromTDengine.class);
25. LOG.info("java.library.path:{}", prop);
26. System.out.println(prop);
27. Class.forName( driver );
28. Properties properties = new Properties();
29. connection = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/tt" + "?user=root&password=taosdata"
30. , properties);
31. statement = connection.createStatement();
32. }
34. @Override
35. public void close() throws Exception {
36. super.close();
37. if (connection != null) {
38. connection.close();
39. }
40. if (statement != null) {
41. statement.close();
42. }
43. }
45. @Override
46. public void run(SourceContext<Sensor> sourceContext) throws Exception {
47. try {
48. String sql = "select * from tt.meters";
49. ResultSet resultSet = statement.executeQuery(sql);
50. while (resultSet.next()) {
51. Sensor sensor = new Sensor( resultSet.getLong(1),
52. resultSet.getInt( "vol" ),
53. resultSet.getFloat( "current" ),
54. resultSet.getString( "location" ).trim());
55. sourceContext.collect( sensor );
56. }
57. } catch (Exception e) {
58. e.printStackTrace();
59. }
60. }
62. @Override
63. public void cancel() {
65. }
66. }

2) 自定義類 SinkToTDengine

用途：資料來源連線，資料寫入

SinkToTDengine

1. package com.taosdata.flink;
3. import org.apache.flink.configuration.Configuration;
4. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
5. import com.taosdata.model.Sensor;
6. import java.sql.*;
7. import java.util.Properties;
10. public class SinkToTDengine extends RichSinkFunction<Sensor> {
11. Statement statement;
12. private Connection connection;
14. @Override
15. public void open(Configuration parameters) throws Exception {
16. super.open(parameters);
17. String driver = "com.taosdata.jdbc.rs.RestfulDriver";
18. String host = "TAOS-FQDN";
19. String username = "root";
20. String password = "taosdata";
21. String prop = System.getProperty("java.library.path");
22. System.out.println(prop);
23. Class.forName( driver );
24. Properties properties = new Properties();
25. connection = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/tt" + "?user=root&password=taosdata"
26. , properties);
27. statement = connection.createStatement();
29. }
31. @Override
32. public void close() throws Exception {
33. super.close();
34. if (connection != null) {
35. connection.close();
36. }
37. if (statement != null) {
38. statement.close();
39. }
40. }
42. @Override
43. public void invoke(Sensor sensor, Context context) throws Exception {
44. try {
45. String sql = String.format("insert into sinktest.%s using sinktest.meters tags('%s') values(%d,%d,%f)",
46. sensor.getLocation(),
47. sensor.getLocation(),
48. sensor.getTs(),
49. sensor.getVal(),
50. sensor.getCurrent()
51. );
52. statement.executeUpdate(sql);
54. } catch (Exception e) {
55. e.printStackTrace();
56. }
57. }
58. }

3) 自定義類 Sensor

用途：定義資料結構，用來接受資料

1. package com.taosdata.model;
3. public class Sensor {
5. public long ts;
6. public int val;
7. public float current;
8. public String location;
10. public Sensor() {
12. }
14. public Sensor(long ts, int val, float current, String location) {
15. this.ts = ts;
16. this.val = val;
17. this.current = current;
18. this.location = location;
19. }
21. public long getTs() {
22. return ts;
23. }
25. public void setTs(long ts) {
26. this.ts = ts;
27. }
29. public int getVal() {
30. return val;
31. }
33. public void setVal(int val) {
34. this.val = val;
35. }
37. public float getCurrent() {
38. return current;
39. }
41. public void setCurrent(float current) {
42. this.current = current;
43. }
45. public String getLocation() {
46. return location;
47. }
49. public void setLocation(String location) {
50. this.location = location;
51. }
53. @Override
54. public String toString() {
55. return "Sensor{" +
56. "ts=" + ts +
57. ", val=" + val +
58. ", current=" + current +
59. ", location='" + location + '\'' +
60. '}';
61. }
62. }

4) 主程式類 ReadFromTDengine

用途：呼叫 Flink 進行讀取和寫入資料

1. package com.taosdata;
3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. import org.apache.flink.api.common.functions.MapFunction;
6. import org.apache.flink.streaming.api.datastream.DataStream;
7. import com.taosdata.model.Sensor;
9. import org.slf4j.LoggerFactory;
10. import org.slf4j.Logger;
12. public class ReadFromTDengine {
13. public static void main(String[] args) throws Exception {
14. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
15. DataStreamSource<Sensor> SensorList = env.addSource( new com.taosdata.flink.SourceFromTDengine() );
16. SensorList.print();
17. SensorList.addSink( new com.taosdata.flink.SinkToTDengine() );
18. env.execute();
20. }
21. }

3、簡單測試 RESTful 介面

1) 環境準備：

a) Flink 安裝&啟動：

• wget https://dlcdn.apache.org/flink/flink-1.14.3/flink-1.14.3-bin-scala_2.12.tgz
• tar zxf flink-1.14.3-bin-scala_2.12.tgz -C /usr/local
• /usr/local/flink-1.14.3/bin/start-cluster.sh

b) TDengine Database 環境準備：

• 建立原始資料： 
  • create database tt;
  • create table `meters` (`ts` TIMESTAMP,`vol` INT,`current` FLOAT) TAGS (`location` BINARY(20));
  • insert into beijing using meters tags(‘beijing’) values(now,220,30.2);
• 建立目標資料庫表： 
  • create database sinktest;
  • create table `meters` (`ts` TIMESTAMP,`vol` INT,`current` FLOAT) TAGS (`location` BINARY(20));

2) 打包編譯：

原始碼位置： https://github.com/liuyq-617/TD-Flink

mvn clean package

3) 程式啟動：

flink run target/test-flink-1.0-SNAPSHOT-dist.jar

• 讀取資料 
  • vi log/flink-root-taskexecutor-0-xxxxx.out 
  • 檢視到資料列印：Sensor{ts=1645166073101, val=220, current=5.7, location=’beijing’}
• 寫入資料 
  • show sinktest.tables; 
    • 已經建立了beijing 子表
  • select * from sinktest.beijing; 
    • 可以查詢到剛插入的資料

4、使用 JNI 方式

舉一反三的小夥伴此時已經猜到，只要把 JDBC URL 修改一下就可以了。

但是 Flink 每次分派作業時都在使用一個新的 ClassLoader，而我們在計算節點上就會得到“Native library already loaded in another classloader”錯誤。

為了避免此問題，可以將 JDBC 的 jar 包放到 Flink 的 lib 目錄下，不去呼叫 dist 包就可以了。

• cp taos-jdbcdriver-2.0.37-dist.jar /usr/local/flink-1.14.3/lib
• flink run target/test-flink-1.0-SNAPSHOT.jar

5、小結

通過在專案中引入 SourceFromTDengine 和 SinkToTDengine 兩個類，即可完成在 Flink 中對 TDengine 的讀寫操作。後面我們會有文章介紹 Spark 和 TDengine 的對接。

注：文中使用的是 JDBC 的 RESTful 介面，這樣就不用在 Flink 的節點安裝 TDengine，JNI 方式需要在 Flink 節點安裝 TDengine Database 的客戶端。

想了解更多 TDengine Database的具體細節，歡迎大家在GitHub上檢視相關原始碼。