InfluxDB 行协议

协议介绍

InfluxDB Line 协议采用一行字符串来表示一行数据。分为四部分：

1. measurement,tag_set field_set timestamp
  • measurement 将作为超级表名。它与 tag_set 之间使用一个英文逗号来分隔。
  • tag_set 将作为标签数据，其格式形如 <tag_key>=<tag_value>,<tag_key>=<tag_value>，也即可以使用英文逗号来分隔多个标签数据。它与 field_set 之间使用一个半角空格来分隔。
  • field_set 将作为普通列数据，其格式形如 <field_key>=<field_value>,<field_key>=<field_value>，同样是使用英文逗号来分隔多个普通列的数据。它与 timestamp 之间使用一个半角空格来分隔。
  • timestamp 即本行数据对应的主键时间戳。

例如：

1. meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611249500
InfluxDB 行协议 - 图1note
  • tag_set 中的所有的数据自动转化为 NCHAR 数据类型;
  • field_set 中的每个数据项都需要对自身的数据类型进行描述, 比如 1.2f32 代表 FLOAT 类型的数值 1.2, 如果不带类型后缀会被当作 DOUBLE 处理;
  • timestamp 支持多种时间精度。写入数据的时候需要用参数指定时间精度，支持从小时到纳秒的 6 种时间精度。
  • 为了提高写入的效率，默认假设同一个超级表中 field_set 的顺序是一样的（第一条数据包含所有的 field，后面的数据按照这个顺序），如果顺序不一样，需要配置参数 smlDataFormat 为 false，否则，数据写入按照相同顺序写入，库中数据会异常。（3.0.1.3 之后的版本 smlDataFormat 默认为 false，从3.0.3.0开始，该配置废弃） TDengine 无模式写入参考指南
  • 默认产生的子表名是根据规则生成的唯一 ID 值。用户也可以通过在 taos.cfg 里配置 smlChildTableName 参数来指定某个标签值作为子表名。该标签值应该具有全局唯一性。举例如下：假设有个标签名为tname, 配置 smlChildTableName=tname, 插入数据为 st,tname=cpu1,t1=4 c1=3 1626006833639000000 则创建的子表名为 cpu1。注意如果多行数据 tname 相同，但是后面的 tag_set 不同，则使用第一行自动建表时指定的 tag_set，其他的行会忽略）。TDengine 无模式写入参考指南

要了解更多可参考：InfluxDB Line 协议官方文档 和 TDengine 无模式写入参考指南

示例代码

  • 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 LineProtocolExample {
13. // format: measurement,tag_set field_set timestamp
14. private static String[] lines = {
15. "meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249000", // micro
16. // seconds
17. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611249500",
18. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249300",
19. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611249800",
20. };

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

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

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

查看源码

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

4. lines = ["meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249000",
5. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611249500",
6. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249300",
7. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611249800",
8. ]

11. def get_connection():
12. # create connection use firstEP in taos.cfg.
13. return taos.connect()

16. def create_database(conn):
17. # the default precision is ms (microsecond), but we use us(microsecond) here.
18. conn.execute("CREATE DATABASE test precision 'us'")
19. conn.execute("USE test")

22. def insert_lines(conn):
23. affected_rows = conn.schemaless_insert(
24. lines, SmlProtocol.LINE_PROTOCOL, SmlPrecision.MICRO_SECONDS)
25. print(affected_rows)  # 8

28. if __name__ == '__main__':
29. connection = get_connection()
30. try:
31. create_database(connection)
32. insert_lines(connection)
33. finally:
34. connection.close()

查看源码

1. package main

3. import (
4. "fmt"
5. "log"

7. "github.com/taosdata/driver-go/v3/af"
8. )

10. func prepareDatabase(conn *af.Connector) {
11. _, err := conn.Exec("CREATE DATABASE test")
12. if err != nil {
13. panic(err)
14. }
15. _, err = conn.Exec("USE test")
16. if err != nil {
17. panic(err)
18. }
19. }

21. func main() {
22. conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
23. if err != nil {
24. fmt.Println("fail to connect, err:", err)
25. }
26. defer conn.Close()
27. prepareDatabase(conn)
28. var lines = []string{
29. "meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
30. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
31. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
32. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250",
33. }

35. err = conn.InfluxDBInsertLines(lines, "ms")
36. if err != nil {
37. log.Fatalln("insert error:", err)
38. }
39. }

查看源码

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,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
17. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
18. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
19. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250",
20. ];
21. cursor.schemalessInsert(
22. lines,
23. taos.SCHEMALESS_PROTOCOL.TSDB_SML_LINE_PROTOCOL,
24. taos.SCHEMALESS_PRECISION.TSDB_SML_TIMESTAMP_MILLI_SECONDS
25. );
26. }

28. try {
29. createDatabase();
30. insertData();
31. } finally {
32. cursor.close();
33. conn.close();
34. }

查看源码

1. using TDengineDriver;

3. namespace TDengineExample
4. {
5. internal class InfluxDBLineExample
6. {
7. static void Main()
8. {
9. IntPtr conn = GetConnection();
10. PrepareDatabase(conn);
11. string[] lines = {
12. "meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
13. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
14. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
15. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250"
16. };
17. IntPtr res = TDengine.SchemalessInsert(conn, lines, lines.Length, (int)TDengineSchemalessProtocol.TSDB_SML_LINE_PROTOCOL, (int)TDengineSchemalessPrecision.TSDB_SML_TIMESTAMP_MILLI_SECONDS);
18. if (TDengine.ErrorNo(res) != 0)
19. {
20. throw new Exception("SchemalessInsert failed since " + TDengine.Error(res));
21. }
22. else
23. {
24. int affectedRows = TDengine.AffectRows(res);
25. Console.WriteLine($"SchemalessInsert success, affected {affectedRows} rows");
26. }
27. TDengine.FreeResult(res);

29. }
30. static IntPtr GetConnection()
31. {
32. string host = "localhost";
33. short port = 6030;
34. string username = "root";
35. string password = "taosdata";
36. string dbname = "";
37. var conn = TDengine.Connect(host, username, password, dbname, port);
38. if (conn == IntPtr.Zero)
39. {
40. throw new Exception("Connect to TDengine failed");
41. }
42. else
43. {
44. Console.WriteLine("Connect to TDengine success");
45. }
46. return conn;
47. }

49. static void PrepareDatabase(IntPtr conn)
50. {
51. IntPtr res = TDengine.Query(conn, "CREATE DATABASE test");
52. if (TDengine.ErrorNo(res) != 0)
53. {
54. throw new Exception("failed to create database, reason: " + TDengine.Error(res));
55. }
56. res = TDengine.Query(conn, "USE test");
57. if (TDengine.ErrorNo(res) != 0)
58. {
59. throw new Exception("failed to change database, reason: " + TDengine.Error(res));
60. }
61. }

63. }

65. }

查看源码

1. int main() {
2. TAOS *taos = taos_connect("localhost", "root", "taosdata", "", 0);
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[] = {"meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
11. "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
12. "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
13. "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250"};
14. TAOS_RES *res = taos_schemaless_insert(taos, lines, 4, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_MILLI_SECONDS);
15. if (taos_errno(res) != 0) {
16. printf("failed to insert schema-less data, reason: %s\n", taos_errstr(res));
17. } else {
18. int affectedRows = taos_affected_rows(res);
19. printf("successfully inserted %d rows\n", affectedRows);
20. }
21. taos_free_result(res);
22. taos_close(taos);
23. taos_cleanup();
24. }
25. // output:
26. // successfully inserted 4 rows

查看源码

SQL 查询示例

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

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

1. SELECT * FROM meters WHERE location = "California.LosAngeles" AND groupid = 2;