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 类型。
• 默认生成的子表名是根据规则生成的唯一 ID 值。用户也可以通过在 taos.cfg 里配置 smlChildTableName 参数来指定某个标签值作为子表名。该标签值应该具有全局唯一性。举例如下：假设有个标签名为tname, 配置 smlChildTableName=tname, 插入数据为 "tags": { "host": "web02","dc": "lga","tname":"cpu1"} 则创建的子表名为 cpu1。注意如果多行数据 tname 相同，但是后面的 tag_set 不同，则使用第一行自动建表时指定的 tag_set，其他的行会忽略）。

示例代码

• Java
• Python
• Go
• 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. "log"

6. "github.com/taosdata/driver-go/v3/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. log.Fatalln("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. log.Fatalln("insert error:", err)
36. }
37. }

查看源码

1. const taos = require("@tdengine/client");

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. try
11. {
12. PrepareDatabase(conn);
13. string[] lines = { "[{\"metric\": \"meters.current\", \"timestamp\": 1648432611249, \"value\": 10.3, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
14. " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611249, \"value\": 219, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}, " +
15. "{\"metric\": \"meters.current\", \"timestamp\": 1648432611250, \"value\": 12.6, \"tags\": {\"location\": \"California.SanFrancisco\", \"groupid\": 2}}," +
16. " {\"metric\": \"meters.voltage\", \"timestamp\": 1648432611250, \"value\": 221, \"tags\": {\"location\": \"California.LosAngeles\", \"groupid\": 1}}]"
17. };

19. IntPtr res = TDengine.SchemalessInsert(conn, lines, 1, (int)TDengineSchemalessProtocol.TSDB_SML_JSON_PROTOCOL, (int)TDengineSchemalessPrecision.TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
20. if (TDengine.ErrorNo(res) != 0)
21. {
22. throw new Exception("SchemalessInsert failed since " + TDengine.Error(res));
23. }
24. else
25. {
26. int affectedRows = TDengine.AffectRows(res);
27. Console.WriteLine($"SchemalessInsert success, affected {affectedRows} rows");
28. }
29. TDengine.FreeResult(res);
30. }
31. finally
32. {
33. TDengine.Close(conn);
34. }
35. }
36. static IntPtr GetConnection()
37. {
38. string host = "localhost";
39. short port = 6030;
40. string username = "root";
41. string password = "taosdata";
42. string dbname = "";
43. var conn = TDengine.Connect(host, username, password, dbname, port);
44. if (conn == IntPtr.Zero)
45. {
46. throw new Exception("Connect to TDengine failed");
47. }
48. else
49. {
50. Console.WriteLine("Connect to TDengine success");
51. }
52. return conn;
53. }

55. static void PrepareDatabase(IntPtr conn)
56. {
57. IntPtr res = TDengine.Query(conn, "CREATE DATABASE test");
58. if (TDengine.ErrorNo(res) != 0)
59. {
60. throw new Exception("failed to create database, reason: " + TDengine.Error(res));
61. }
62. res = TDengine.Query(conn, "USE test");
63. if (TDengine.ErrorNo(res) != 0)
64. {
65. throw new Exception("failed to change database, reason: " + TDengine.Error(res));
66. }
67. }
68. }
69. }

查看源码

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              |
6. =================================
7. meters.current                 |
8. meters.voltage                 |
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)

SQL 查询示例

meters.voltage 是插入数据的超级表名。

可以通过超级表的 TAG 来过滤数据，比如查询 location=California.LosAngeles groupid=1 可以通过如下 SQL：

1. SELECT * FROM `meters.current` WHERE location = "California.LosAngeles" AND groupid = 3;