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
示例代码
- 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)