OpenTSDB 行协议

协议介绍

OpenTSDB 行协议同样采用一行字符串来表示一行数据。OpenTSDB 采用的是单列模型，因此一行只能包含一个普通数据列。标签列依然可以有多个。分为四部分，具体格式约定如下：

1. <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
  • metric 将作为超级表名；
  • timestamp 本行数据对应的时间戳。根据时间戳的长度自动识别时间精度。支持秒和毫秒两种时间精度；
  • value 度量值，必须为一个数值。对应的列名是 “_value”；
  • 最后一部分是标签集， 用空格分隔不同标签， 所有标签自动转化为 NCHAR 数据类型。

例如：

1. meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3
  • 默认生产的子表名是根据规则生成的唯一 ID 值。用户也可以通过在 taos.cfg 里配置 smlChildTableName 参数来指定某个标签值作为子表名。该标签值应该具有全局唯一性。举例如下：假设有个标签名为tname, 配置 smlChildTableName=tname, 插入数据为 meters.current 1648432611250 11.3 tname=cpu1 location=California.LosAngeles groupid=3 则创建的表名为 cpu1，注意如果多行数据 tname 相同，但是后面的 tag_set 不同，则使用第一行自动建表时指定的 tag_set，其他的行会忽略）。 参考 OpenTSDB Telnet API 文档。

示例代码

  • 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 TelnetLineProtocolExample {
13. // format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
14. private static String[] lines = { "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
15. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
16. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
17. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
18. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
19. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
20. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
21. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
22. };

24. private static Connection getConnection() throws SQLException {
25. String jdbcUrl = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
26. return DriverManager.getConnection(jdbcUrl);
27. }

29. private static void createDatabase(Connection conn) throws SQLException {
30. try (Statement stmt = conn.createStatement()) {
31. // the default precision is ms (microsecond), but we use us(microsecond) here.
32. stmt.execute("CREATE DATABASE IF NOT EXISTS test precision 'us'");
33. stmt.execute("USE test");
34. }
35. }

37. public static void main(String[] args) throws SQLException {
38. try (Connection conn = getConnection()) {
39. createDatabase(conn);
40. SchemalessWriter writer = new SchemalessWriter(conn);
41. writer.write(lines, SchemalessProtocolType.TELNET, SchemalessTimestampType.NOT_CONFIGURED);
42. }
43. }

45. }

查看源码

1. import taos
2. from taos import SmlProtocol, SmlPrecision

4. # format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
5. lines = ["meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
6. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
7. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
8. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
9. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
10. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
11. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
12. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
13. ]

16. # create connection use firstEp in taos.cfg.
17. def get_connection():
18. return taos.connect()

21. def create_database(conn):
22. conn.execute("CREATE DATABASE test")
23. conn.execute("USE test")

26. def insert_lines(conn):
27. affected_rows = conn.schemaless_insert(
28. lines, SmlProtocol.TELNET_PROTOCOL, SmlPrecision.NOT_CONFIGURED)
29. print(affected_rows)  # 8

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)
27. var lines = []string{
28. "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
29. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
30. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
31. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
32. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
33. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
34. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
35. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
36. }

38. err = conn.OpenTSDBInsertTelnetLines(lines)
39. if err != nil {
40. log.Fatalln("insert error:", err)
41. }
42. }

查看源码

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. "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
17. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
18. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
19. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
20. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
21. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
22. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
23. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
24. ];
25. cursor.schemalessInsert(
26. lines,
27. taos.SCHEMALESS_PROTOCOL.TSDB_SML_TELNET_PROTOCOL,
28. taos.SCHEMALESS_PRECISION.TSDB_SML_TIMESTAMP_NOT_CONFIGURED
29. );
30. }

32. try {
33. createDatabase();
34. insertData();
35. } finally {
36. cursor.close();
37. conn.close();
38. }

查看源码

1. using TDengineDriver;

3. namespace TDengineExample
4. {
5. internal class OptsTelnetExample
6. {
7. static void Main()
8. {
9. IntPtr conn = GetConnection();
10. try
11. {
12. PrepareDatabase(conn);
13. string[] lines = {
14. "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
15. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
16. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
17. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
18. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
19. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
20. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
21. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
22. };
23. IntPtr res = TDengine.SchemalessInsert(conn, lines, lines.Length, (int)TDengineSchemalessProtocol.TSDB_SML_TELNET_PROTOCOL, (int)TDengineSchemalessPrecision.TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
24. if (TDengine.ErrorNo(res) != 0)
25. {
26. throw new Exception("SchemalessInsert failed since " + TDengine.Error(res));
27. }
28. else
29. {
30. int affectedRows = TDengine.AffectRows(res);
31. Console.WriteLine($"SchemalessInsert success, affected {affectedRows} rows");
32. }
33. TDengine.FreeResult(res);
34. }
35. catch
36. {
37. TDengine.Close(conn);
38. }
39. }
40. static IntPtr GetConnection()
41. {
42. string host = "localhost";
43. short port = 6030;
44. string username = "root";
45. string password = "taosdata";
46. string dbname = "";
47. var conn = TDengine.Connect(host, username, password, dbname, port);
48. if (conn == IntPtr.Zero)
49. {
50. throw new Exception("Connect to TDengine failed");
51. }
52. else
53. {
54. Console.WriteLine("Connect to TDengine success");
55. }
56. return conn;
57. }

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

查看源码

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 *lines[] = {
11. "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
12. "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
13. "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
14. "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
15. "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
16. "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
17. "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
18. "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
19. };
20. TAOS_RES *res = taos_schemaless_insert(taos, lines, 8, TSDB_SML_TELNET_PROTOCOL, TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
21. if (taos_errno(res) != 0) {
22. printf("failed to insert schema-less data, reason: %s\n", taos_errstr(res));
23. } else {
24. int affectedRow = taos_affected_rows(res);
25. printf("successfully inserted %d rows\n", affectedRow);
26. }

28. taos_free_result(res);
29. taos_close(taos);
30. taos_cleanup();
31. }
32. // output:
33. // successfully inserted 8 rows

查看源码

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

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.002544s)

11. taos> SELECT TBNAME, * FROM `meters.current`;
12. tbname             |           _ts            |           _value           | groupid |            location            |
13. ==================================================================================================================================
14. t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.249 |              10.800000000 | 3       | California.LosAngeles                |
15. t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.250 |              11.300000000 | 3       | California.LosAngeles                |
16. t_7e7b26dd860280242c6492a16... | 2022-03-28 09:56:51.249 |              10.300000000 | 2       | California.SanFrancisco               |
17. t_7e7b26dd860280242c6492a16... | 2022-03-28 09:56:51.250 |              12.600000000 | 2       | California.SanFrancisco               |
18. Query OK, 4 row(s) in set (0.005399s)

SQL 查询示例

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

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

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