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 种时间精度。

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

示例代码

  • 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 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"
  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)
  27.     var lines = []string{
  28.         "meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
  29.         "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
  30.         "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
  31.         "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250",
  32.     }
  34.     err = conn.InfluxDBInsertLines(lines, "ms")
  35.     if err != nil {
  36.         fmt.Println("insert error:", err)
  37.     }
  38. }

查看源码

  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 = ["meters,location=California.LosAngeles,groupid=2 current=11.8,voltage=221,phase=0.28 1648432611249",
  9.     "meters,location=California.LosAngeles,groupid=2 current=13.4,voltage=223,phase=0.29 1648432611250",
  10.     "meters,location=California.LosAngeles,groupid=3 current=10.8,voltage=223,phase=0.29 1648432611249",
  11.     "meters,location=California.LosAngeles,groupid=3 current=11.3,voltage=221,phase=0.35 1648432611250"];
  12.     let affected_rows = taos
  13.         .schemaless_insert(
  14.             &lines,
  15.             TSDB_SML_LINE_PROTOCOL,
  16.             TSDB_SML_TIMESTAMP_MILLISECONDS,
  17.         )
  18.         .unwrap();
  19.     println!("affected_rows={}", affected_rows);
  20. }
  22. // run with:  cargo run --example influxdb_line_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.     "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.                 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.     }
  77. }

查看源码

  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

查看源码