OpenTSDB JSON 格式协议

协议介绍

OpenTSDB JSON 格式协议采用一个 JSON 字符串表示一行或多行数据。例如：

  1. [
  2.   {
  3.     "metric": "sys.cpu.nice",
  4.     "timestamp": 1346846400,
  5.     "value": 18,
  6.     "tags": {
  7.       "host": "web01",
  8.       "dc": "lga"
  9.     }
  10.   },
  11.   {
  12.     "metric": "sys.cpu.nice",
  13.     "timestamp": 1346846400,
  14.     "value": 9,
  15.     "tags": {
  16.       "host": "web02",
  17.       "dc": "lga"
  18.     }
  19.   }
  20. ]

与 OpenTSDB 行协议类似， metric 将作为超级表名， timestamp 表示时间戳，value 表示度量值， tags 表示标签集。

参考OpenTSDB HTTP API 文档。

OpenTSDB JSON 格式协议 - 图1note
  • 对于 JSON 格式协议，TDengine 并不会自动把所有标签转成 nchar 类型， 字符串将将转为 nchar 类型， 数值将同样转换为 double 类型。
  • TDengine 只接收 JSON 数组格式的字符串，即使一行数据也需要转换成数组形式。

示例代码

  • Java
  • Python
  • Go
  • Rust
  • Node.js
  • C#
  • C
  1. package com.taos.example;
  3. import com.taosdata.jdbc.SchemalessWriter;
  4. import com.taosdata.jdbc.enums.SchemalessProtocolType;
  5. import com.taosdata.jdbc.enums.SchemalessTimestampType;
  7. import java.sql.Connection;
  8. import java.sql.DriverManager;
  9. import java.sql.SQLException;
  10. import java.sql.Statement;
  12. public class JSONProtocolExample {
  13.     private static Connection getConnection() throws SQLException {
  14.         String jdbcUrl = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
  15.         return DriverManager.getConnection(jdbcUrl);
  16.     }
  18.     private static void createDatabase(Connection conn) throws SQLException {
  19.         try (Statement stmt = conn.createStatement()) {
  20.             stmt.execute("CREATE DATABASE IF NOT EXISTS test");
  21.             stmt.execute("USE test");
  22.         }
  23.     }
  25.     private static String getJSONData() {
  26.         return "[{\"metric\": \"meters.current\", \"timestamp\": 1648432611249, \"value\": 10.3, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
  27.                 " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611249, \"value\": 219, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}, " +
  28.                 "{\"metric\": \"meters.current\", \"timestamp\": 1648432611250, \"value\": 12.6, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
  29.                 " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611250, \"value\": 221, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}]";
  30.     }
  32.     public static void main(String[] args) throws SQLException {
  33.         try (Connection conn = getConnection()) {
  34.             createDatabase(conn);
  35.             SchemalessWriter writer = new SchemalessWriter(conn);
  36.             String jsonData = getJSONData();
  37.             writer.write(jsonData, SchemalessProtocolType.JSON, SchemalessTimestampType.NOT_CONFIGURED);
  38.         }
  39.     }
  40. }

查看源码

  1. import json
  3. import taos
  4. from taos import SmlProtocol, SmlPrecision
  6. lines = [{"metric": "meters.current", "timestamp": 1648432611249, "value": 10.3, "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  7.          {"metric": "meters.voltage", "timestamp": 1648432611249, "value": 219,
  8.           "tags": {"location": "California.LosAngeles", "groupid": 1}},
  9.          {"metric": "meters.current", "timestamp": 1648432611250, "value": 12.6,
  10.           "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  11.          {"metric": "meters.voltage", "timestamp": 1648432611250, "value": 221, "tags": {"location": "California.LosAngeles", "groupid": 1}}]
  14. def get_connection():
  15.     return taos.connect()
  18. def create_database(conn):
  19.     conn.execute("CREATE DATABASE test")
  20.     conn.execute("USE test")
  23. def insert_lines(conn):
  24.     global lines
  25.     lines = json.dumps(lines)
  26.     # note: the first parameter must be a list with only one element.
  27.     affected_rows = conn.schemaless_insert(
  28.         [lines], SmlProtocol.JSON_PROTOCOL, SmlPrecision.NOT_CONFIGURED)
  29.     print(affected_rows)  # 4
  32. if __name__ == '__main__':
  33.     connection = get_connection()
  34.     try:
  35.         create_database(connection)
  36.         insert_lines(connection)
  37.     finally:
  38.         connection.close()

查看源码

  1. package main
  3. import (
  4.     "fmt"
  6.     "github.com/taosdata/driver-go/v2/af"
  7. )
  9. func prepareDatabase(conn *af.Connector) {
  10.     _, err := conn.Exec("CREATE DATABASE test")
  11.     if err != nil {
  12.         panic(err)
  13.     }
  14.     _, err = conn.Exec("USE test")
  15.     if err != nil {
  16.         panic(err)
  17.     }
  18. }
  20. func main() {
  21.     conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
  22.     if err != nil {
  23.         fmt.Println("fail to connect, err:", err)
  24.     }
  25.     defer conn.Close()
  26.     prepareDatabase(conn)
  28.     payload := `[{"metric": "meters.current", "timestamp": 1648432611249, "value": 10.3, "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  29.                 {"metric": "meters.voltage", "timestamp": 1648432611249, "value": 219, "tags": {"location": "California.LosAngeles", "groupid": 1}},
  30.                 {"metric": "meters.current", "timestamp": 1648432611250, "value": 12.6, "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  31.                 {"metric": "meters.voltage", "timestamp": 1648432611250, "value": 221, "tags": {"location": "California.LosAngeles", "groupid": 1}}]`
  33.     err = conn.OpenTSDBInsertJsonPayload(payload)
  34.     if err != nil {
  35.         fmt.Println("insert error:", err)
  36.     }
  37. }

查看源码

  1. use libtaos::schemaless::*;
  2. use libtaos::*;
  4. fn main() {
  5.     let taos = TaosCfg::default().connect().expect("fail to connect");
  6.     taos.raw_query("CREATE DATABASE test").unwrap();
  7.     taos.raw_query("USE test").unwrap();
  8.     let lines = [
  9.         r#"[{"metric": "meters.current", "timestamp": 1648432611249, "value": 10.3, "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  10.         {"metric": "meters.voltage", "timestamp": 1648432611249, "value": 219, "tags": {"location": "California.LosAngeles", "groupid": 1}},
  11.         {"metric": "meters.current", "timestamp": 1648432611250, "value": 12.6, "tags": {"location": "California.SanFrancisco", "groupid": 2}},
  12.         {"metric": "meters.voltage", "timestamp": 1648432611250, "value": 221, "tags": {"location": "California.LosAngeles", "groupid": 1}}]"#,
  13.     ];
  15.     let affected_rows = taos
  16.         .schemaless_insert(
  17.             &lines,
  18.             TSDB_SML_JSON_PROTOCOL,
  19.             TSDB_SML_TIMESTAMP_NOT_CONFIGURED,
  20.         )
  21.         .unwrap();
  22.     println!("affected_rows={}", affected_rows); // affected_rows=4
  23. }
  25. // run with:  cargo run --example opentsdb_json_example

查看源码

  1. const taos = require("td2.0-connector");
  3. const conn = taos.connect({
  4.   host: "localhost",
  5. });
  7. const cursor = conn.cursor();
  9. function createDatabase() {
  10.   cursor.execute("CREATE DATABASE test");
  11.   cursor.execute("USE test");
  12. }
  14. function insertData() {
  15.   const lines = [
  16.     {
  17.       metric: "meters.current",
  18.       timestamp: 1648432611249,
  19.       value: 10.3,
  20.       tags: { location: "California.SanFrancisco", groupid: 2 },
  21.     },
  22.     {
  23.       metric: "meters.voltage",
  24.       timestamp: 1648432611249,
  25.       value: 219,
  26.       tags: { location: "California.LosAngeles", groupid: 1 },
  27.     },
  28.     {
  29.       metric: "meters.current",
  30.       timestamp: 1648432611250,
  31.       value: 12.6,
  32.       tags: { location: "California.SanFrancisco", groupid: 2 },
  33.     },
  34.     {
  35.       metric: "meters.voltage",
  36.       timestamp: 1648432611250,
  37.       value: 221,
  38.       tags: { location: "California.LosAngeles", groupid: 1 },
  39.     },
  40.   ];
  42.   cursor.schemalessInsert(
  43.     [JSON.stringify(lines)],
  44.     taos.SCHEMALESS_PROTOCOL.TSDB_SML_JSON_PROTOCOL,
  45.     taos.SCHEMALESS_PRECISION.TSDB_SML_TIMESTAMP_NOT_CONFIGURED
  46.   );
  47. }
  49. try {
  50.   createDatabase();
  51.   insertData();
  52. } finally {
  53.   cursor.close(); 
  54.   conn.close();
  55. }

查看源码

  1. using TDengineDriver;
  3. namespace TDengineExample
  4. {
  5.     internal class OptsJsonExample
  6.     {
  7.         static void Main()
  8.         {
  9.             IntPtr conn = GetConnection();
  10.             PrepareDatabase(conn);
  11.             string[] lines = { "[{\"metric\": \"meters.current\", \"timestamp\": 1648432611249, \"value\": 10.3, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
  12.                 " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611249, \"value\": 219, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}, " +
  13.                 "{\"metric\": \"meters.current\", \"timestamp\": 1648432611250, \"value\": 12.6, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
  14.                 " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611250, \"value\": 221, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}]"
  15.             };
  17.             IntPtr res = TDengine.SchemalessInsert(conn, lines, 1, (int)TDengineSchemalessProtocol.TSDB_SML_JSON_PROTOCOL, (int)TDengineSchemalessPrecision.TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
  18.             if (TDengine.ErrorNo(res) != 0)
  19.             {
  20.                 Console.WriteLine("SchemalessInsert failed since " + TDengine.Error(res));
  21.                 ExitProgram(conn, 1);
  22.             }
  23.             else
  24.             {
  25.                 int affectedRows = TDengine.AffectRows(res);
  26.                 Console.WriteLine($"SchemalessInsert success, affected {affectedRows} rows");
  27.             }
  28.             TDengine.FreeResult(res);
  29.             ExitProgram(conn, 0);
  31.         }
  32.         static IntPtr GetConnection()
  33.         {
  34.             string host = "localhost";
  35.             short port = 6030;
  36.             string username = "root";
  37.             string password = "taosdata";
  38.             string dbname = "";
  39.             var conn = TDengine.Connect(host, username, password, dbname, port);
  40.             if (conn == IntPtr.Zero)
  41.             {
  42.                 Console.WriteLine("Connect to TDengine failed");
  43.                 TDengine.Cleanup();
  44.                 Environment.Exit(1);
  45.             }
  46.             else
  47.             {
  48.                 Console.WriteLine("Connect to TDengine success");
  49.             }
  50.             return conn;
  51.         }
  53.         static void PrepareDatabase(IntPtr conn)
  54.         {
  55.             IntPtr res = TDengine.Query(conn, "CREATE DATABASE test");
  56.             if (TDengine.ErrorNo(res) != 0)
  57.             {
  58.                 Console.WriteLine("failed to create database, reason: " + TDengine.Error(res));
  59.                 ExitProgram(conn, 1);
  60.             }
  61.             res = TDengine.Query(conn, "USE test");
  62.             if (TDengine.ErrorNo(res) != 0)
  63.             {
  64.                 Console.WriteLine("failed to change database, reason: " + TDengine.Error(res));
  65.                 ExitProgram(conn, 1);
  66.             }
  67.         }
  69.         static void ExitProgram(IntPtr conn, int exitCode)
  70.         {
  71.             TDengine.Close(conn);
  72.             TDengine.Cleanup();
  73.             Environment.Exit(exitCode);
  74.         }
  75.     }
  76. }

查看源码

  1. int main() {
  2.   TAOS *taos = taos_connect("localhost", "root", "taosdata", "", 6030);
  3.   if (taos == NULL) {
  4.     printf("failed to connect to server\n");
  5.     exit(EXIT_FAILURE);
  6.   }
  7.   executeSQL(taos, "DROP DATABASE IF EXISTS test");
  8.   executeSQL(taos, "CREATE DATABASE test");
  9.   executeSQL(taos, "USE test");
  10.   char *line =
  11.       "[{\"metric\": \"meters.current\", \"timestamp\": 1648432611249, \"value\": 10.3, \"tags\": {\"location\": "
  12.       "\"California.SanFrancisco\", \"groupid\": 2}},{\"metric\": \"meters.voltage\", \"timestamp\": 1648432611249, "
  13.       "\"value\": 219, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}},{\"metric\": \"meters.current\", "
  14.       "\"timestamp\": 1648432611250, \"value\": 12.6, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": "
  15.       "2}},{\"metric\": \"meters.voltage\", \"timestamp\": 1648432611250, \"value\": 221, \"tags\": {\"location\": "
  16.       "\"California.LosAngeles\", \"groupid\": 1}}]";
  18.   char     *lines[] = {line};
  19.   TAOS_RES *res = taos_schemaless_insert(taos, lines, 1, TSDB_SML_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
  20.   if (taos_errno(res) != 0) {
  21.     printf("failed to insert schema-less data, reason: %s\n", taos_errstr(res));
  22.   } else {
  23.     int affectedRow = taos_affected_rows(res);
  24.     printf("successfully inserted %d rows\n", affectedRow);
  25.   }
  26.   taos_free_result(res);
  27.   taos_close(taos);
  28.   taos_cleanup();
  29. }
  30. // output:
  31. // successfully inserted 4 rows

查看源码

以上示例代码会自动创建 2 个超级表， 每个超级表有 2 条数据。

  1. taos> use test;
  2. Database changed.
  4. taos> show stables;
  5.               name              |      created_time       | columns |  tags  |   tables    |
  6. ============================================================================================
  7.  meters.current                 | 2022-03-29 16:05:25.193 |       2 |      2 |           1 |
  8.  meters.voltage                 | 2022-03-29 16:05:25.200 |       2 |      2 |           1 |
  9. Query OK, 2 row(s) in set (0.001954s)
  11. taos> select * from `meters.current`;
  12.            ts            |           value           |          groupid          |            location            |
  13. ===================================================================================================================
  14.  2022-03-28 09:56:51.249 |              10.300000000 |               2.000000000 | California.SanFrancisco               |
  15.  2022-03-28 09:56:51.250 |              12.600000000 |               2.000000000 | California.SanFrancisco               |
  16. Query OK, 2 row(s) in set (0.004076s)