OpenTSDB 行协议

协议介绍

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

<metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]

• metric 将作为超级表名。
• timestamp 本行数据对应的时间戳。根据时间戳的长度自动识别时间精度。支持秒和毫秒两种时间精度
• value 度量值，必须为一个数值。对应的列名也是 “value”。
• 最后一部分是标签集， 用空格分隔不同标签， 所有标签自动转化为 nchar 数据类型;

例如：

meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3

参考OpenTSDB Telnet API 文档。

示例代码

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C

package com.taos.example;
import com.taosdata.jdbc.SchemalessWriter;
import com.taosdata.jdbc.enums.SchemalessProtocolType;
import com.taosdata.jdbc.enums.SchemalessTimestampType;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
public class TelnetLineProtocolExample {
    // format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
    private static String[] lines = { "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
            "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
            "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
            "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
            "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
            "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
            "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
            "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
    };
    private static Connection getConnection() throws SQLException {
        String jdbcUrl = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
        return DriverManager.getConnection(jdbcUrl);
    }
    private static void createDatabase(Connection conn) throws SQLException {
        try (Statement stmt = conn.createStatement()) {
            // the default precision is ms (microsecond), but we use us(microsecond) here.
            stmt.execute("CREATE DATABASE IF NOT EXISTS test precision 'us'");
            stmt.execute("USE test");
        }
    }
    public static void main(String[] args) throws SQLException {
        try (Connection conn = getConnection()) {
            createDatabase(conn);
            SchemalessWriter writer = new SchemalessWriter(conn);
            writer.write(lines, SchemalessProtocolType.TELNET, SchemalessTimestampType.NOT_CONFIGURED);
        }
    }
}

查看源码

import taos
from taos import SmlProtocol, SmlPrecision
# format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
lines = ["meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
         "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
         "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
         "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
         "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
         "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
         "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
         "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
         ]
# create connection use firstEp in taos.cfg.
def get_connection():
    return taos.connect()
def create_database(conn):
    conn.execute("CREATE DATABASE test")
    conn.execute("USE test")
def insert_lines(conn):
    affected_rows = conn.schemaless_insert(
        lines, SmlProtocol.TELNET_PROTOCOL, SmlPrecision.NOT_CONFIGURED)
    print(affected_rows)  # 8
if __name__ == '__main__':
    connection = get_connection()
    try:
        create_database(connection)
        insert_lines(connection)
    finally:
        connection.close()

查看源码

package main
import (
    "fmt"
    "github.com/taosdata/driver-go/v2/af"
)
func prepareDatabase(conn *af.Connector) {
    _, err := conn.Exec("CREATE DATABASE test")
    if err != nil {
        panic(err)
    }
    _, err = conn.Exec("USE test")
    if err != nil {
        panic(err)
    }
}
func main() {
    conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
    if err != nil {
        fmt.Println("fail to connect, err:", err)
    }
    defer conn.Close()
    prepareDatabase(conn)
    var lines = []string{
        "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
        "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
        "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
        "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
        "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
        "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
        "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
        "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
    }
    err = conn.OpenTSDBInsertTelnetLines(lines)
    if err != nil {
        fmt.Println("insert error:", err)
    }
}

查看源码

use libtaos::schemaless::*;
use libtaos::*;
fn main() {
    let taos = TaosCfg::default().connect().expect("fail to connect");
    taos.raw_query("CREATE DATABASE test").unwrap();
    taos.raw_query("USE test").unwrap();
    let lines = [
        "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
        "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
        "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
        "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
        "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
        "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
        "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
        "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
    ];
    let affected_rows = taos
        .schemaless_insert(
            &lines,
            TSDB_SML_TELNET_PROTOCOL,
            TSDB_SML_TIMESTAMP_NOT_CONFIGURED,
        )
        .unwrap();
    println!("affected_rows={}", affected_rows); // affected_rows=8
}
// run with:  cargo run --example opentsdb_telnet_example

查看源码

const taos = require("td2.0-connector");
const conn = taos.connect({
  host: "localhost",
});
const cursor = conn.cursor();
function createDatabase() {
  cursor.execute("CREATE DATABASE test");
  cursor.execute("USE test");
}
function insertData() {
  const lines = [
    "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
    "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
    "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
    "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
    "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
    "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
    "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
    "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
  ];
  cursor.schemalessInsert(
    lines,
    taos.SCHEMALESS_PROTOCOL.TSDB_SML_TELNET_PROTOCOL,
    taos.SCHEMALESS_PRECISION.TSDB_SML_TIMESTAMP_NOT_CONFIGURED
  );
}
try {
  createDatabase();
  insertData();
} finally {
  cursor.close();
  conn.close();
}

查看源码

using TDengineDriver;
namespace TDengineExample
{
    internal class OptsTelnetExample
    {
        static void Main()
        {
            IntPtr conn = GetConnection();
            PrepareDatabase(conn);
            string[] lines = {
                "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
                "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
                "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
                "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
                "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
                "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
                "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
                "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
            };
            IntPtr res = TDengine.SchemalessInsert(conn, lines, lines.Length, (int)TDengineSchemalessProtocol.TSDB_SML_TELNET_PROTOCOL, (int)TDengineSchemalessPrecision.TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
            if (TDengine.ErrorNo(res) != 0)
            {
                Console.WriteLine("SchemalessInsert failed since " + TDengine.Error(res));
                ExitProgram(conn, 1);
            }
            else
            {
                int affectedRows = TDengine.AffectRows(res);
                Console.WriteLine($"SchemalessInsert success, affected {affectedRows} rows");
            }
            TDengine.FreeResult(res);
            ExitProgram(conn, 0);
        }
        static IntPtr GetConnection()
        {
            string host = "localhost";
            short port = 6030;
            string username = "root";
            string password = "taosdata";
            string dbname = "";
            var conn = TDengine.Connect(host, username, password, dbname, port);
            if (conn == IntPtr.Zero)
            {
                Console.WriteLine("Connect to TDengine failed");
                TDengine.Cleanup();
                Environment.Exit(1);
            }
            else
            {
                Console.WriteLine("Connect to TDengine success");
            }
            return conn;
        }
        static void PrepareDatabase(IntPtr conn)
        {
            IntPtr res = TDengine.Query(conn, "CREATE DATABASE test");
            if (TDengine.ErrorNo(res) != 0)
            {
                Console.WriteLine("failed to create database, reason: " + TDengine.Error(res));
                ExitProgram(conn, 1);
            }
            res = TDengine.Query(conn, "USE test");
            if (TDengine.ErrorNo(res) != 0)
            {
                Console.WriteLine("failed to change database, reason: " + TDengine.Error(res));
                ExitProgram(conn, 1);
            }
        }
        static void ExitProgram(IntPtr conn, int exitCode)
        {
            TDengine.Close(conn);
            TDengine.Cleanup();
            Environment.Exit(exitCode);
        }
    }
}

查看源码

int main() {
  TAOS *taos = taos_connect("localhost", "root", "taosdata", "", 6030);
  if (taos == NULL) {
    printf("failed to connect to server\n");
    exit(EXIT_FAILURE);
  }
  executeSQL(taos, "DROP DATABASE IF EXISTS test");
  executeSQL(taos, "CREATE DATABASE test");
  executeSQL(taos, "USE test");
  char *lines[] = {
      "meters.current 1648432611249 10.3 location=California.SanFrancisco groupid=2",
      "meters.current 1648432611250 12.6 location=California.SanFrancisco groupid=2",
      "meters.current 1648432611249 10.8 location=California.LosAngeles groupid=3",
      "meters.current 1648432611250 11.3 location=California.LosAngeles groupid=3",
      "meters.voltage 1648432611249 219 location=California.SanFrancisco groupid=2",
      "meters.voltage 1648432611250 218 location=California.SanFrancisco groupid=2",
      "meters.voltage 1648432611249 221 location=California.LosAngeles groupid=3",
      "meters.voltage 1648432611250 217 location=California.LosAngeles groupid=3",
  };
  TAOS_RES *res = taos_schemaless_insert(taos, lines, 8, TSDB_SML_TELNET_PROTOCOL, TSDB_SML_TIMESTAMP_NOT_CONFIGURED);
  if (taos_errno(res) != 0) {
    printf("failed to insert schema-less data, reason: %s\n", taos_errstr(res));
  } else {
    int affectedRow = taos_affected_rows(res);
    printf("successfully inserted %d rows\n", affectedRow);
  }
  taos_free_result(res);
  taos_close(taos);
  taos_cleanup();
}
// output:
// successfully inserted 8 rows

查看源码

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

taos> use test;
Database changed.
taos> show stables;
              name              |      created_time       | columns |  tags  |   tables    |
============================================================================================
 meters.current                 | 2022-03-30 17:04:10.877 |       2 |      2 |           2 |
 meters.voltage                 | 2022-03-30 17:04:10.882 |       2 |      2 |           2 |
Query OK, 2 row(s) in set (0.002544s)
taos> select tbname, * from `meters.current`;
             tbname             |           ts            |           value           | groupid |            location            |
==================================================================================================================================
 t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.249 |              10.800000000 | 3       | California.LosAngeles                |
 t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.250 |              11.300000000 | 3       | California.LosAngeles                |
 t_7e7b26dd860280242c6492a16... | 2022-03-28 09:56:51.249 |              10.300000000 | 2       | California.SanFrancisco               |
 t_7e7b26dd860280242c6492a16... | 2022-03-28 09:56:51.250 |              12.600000000 | 2       | California.SanFrancisco               |
Query OK, 4 row(s) in set (0.005399s)