Apache IoTDB Rust客户端 v0.3.3 正式版发布
Apache IoTDB Rust 客户端 v0.3.3 正式版发布，同时也是第一个稳定版本，更新功能：
时区支持
- get_time_zone
- set_time_zone
存储组
- set_storage_group
- delete_storage_group
时间序列
- create_timeseries
- create_multi_timeseries
- delete_timeseries
数据写入
- insert_record
- insert_records_of_one_device
- insert_records
- insert_string_record
- insert_tablet
- insert_tablets
- delete_data
查询
- execute_statement
- execute_query_statemen
- execute_batch_statement
- execute_raw_data_query
- execute_update_statement
备注：iotdb-client-rs 贡献者 https://github.com/manlge ，同时也是 iotdb-client-go 的官方客户端贡献者
Apache IoTDB Client for Rust
概述
Apache IoTDB Rust 客户端
Apache IoTDB 官网: https://iotdb.apache.org
Apache IoTDB Github 仓库: https://github.com/apache/iotdb
预置项
apache-iotdb 0.12.0 或更新版本
快速开始使用：
使用
Put this in your Cargo.toml:
[dependencies]
iotdb-client-rs="0.3.3"
chrono="0.4.19"
prettytable-rs="0.8.0"
use std::vec;
use chrono;
use chrono::Local;
use iotdb_client_rs::client::remote::{Config, RpcSession};
use iotdb_client_rs::client::{MeasurementSchema, Result, RowRecord, Session, Tablet, Value};
use iotdb_client_rs::protocal::{TSCompressionType, TSDataType, TSEncoding};
use prettytable::{cell, Row, Table};
fn main() {
run().expect("failed to run session_example.")
}
fn run() -> Result<()> {
let config = Config {
host: String::from("127.0.0.1"),
port: 6667,
username: String::from("root"),
password: String::from("root"),
..Default::default()
};
let mut session = RpcSession::new(&config)?;
session.open()?;
let tz = session.get_time_zone()?;
if tz != "Asia/Shanghai" {
session.set_time_zone("Asia/Shanghai")?;
}
session.set_storage_group("root.ln1")?;
session.delete_storage_group("root.ln1")?;
session.set_storage_group("root.ln1")?;
session.set_storage_group("root.ln2")?;
session.delete_storage_groups(vec!["root.ln1", "root.ln2"])?;
session.create_timeseries(
"root.sg_rs.dev2.status",
TSDataType::Float,
TSEncoding::Plain,
TSCompressionType::SNAPPY,
None,
None,
None,
None,
)?;
session.delete_timeseries(vec!["root.sg_rs.dev2.status"])?;
session.insert_record(
"root.sg_rs.dev5",
vec!["online", "desc"],
vec![Value::Bool(false), Value::Text("F4145".to_string())],
Local::now().timestamp_millis(),
false,
)?;
session.delete_timeseries(vec!["root.sg_rs.dev5.online", "root.sg_rs.dev5.desc"])?;
session.insert_records(
vec!["root.sg_rs.dev1"],
vec![vec![
"restart_count",
"tick_count",
"price",
"temperature",
"description",
"status",
]],
vec![vec![
Value::Int32(1),
Value::Int64(2018),
Value::Double(1988.1),
Value::Float(12.1),
Value::Text("Test Device 1".to_string()),
Value::Bool(false),
]],
vec![Local::now().timestamp_millis()],
)?;
session.delete_timeseries(vec![
"root.sg_rs.dev1.restart_count",
"root.sg_rs.dev1.tick_count",
"root.sg_rs.dev1.price",
"root.sg_rs.dev1.temperature",
"root.sg_rs.dev1.description",
"root.sg_rs.dev1.status",
])?;
session.create_multi_timeseries(
vec!["root.sg3.dev1.temperature", "root.sg3.dev1.desc"],
vec![TSDataType::Float, TSDataType::Text],
vec![TSEncoding::Plain, TSEncoding::Plain],
vec![TSCompressionType::SNAPPY, TSCompressionType::SNAPPY],
None,
None,
None,
None,
)?;
session.delete_timeseries(vec!["root.sg3.dev1.temperature", "root.sg3.dev1.desc"])?;
session.insert_string_record(
"root.sg_rs.wf02.wt02",
vec!["id", "location"],
vec!["SN:001", "BeiJing"],
Local::now().timestamp_millis(),
false,
)?;
session.delete_timeseries(vec![
"root.sg_rs.wf02.wt02.id",
"root.sg_rs.wf02.wt02.location",
])?;
session.insert_records_of_one_device(
"root.sg_rs.dev0",
vec![
Local::now().timestamp_millis(),
Local::now().timestamp_millis() - 1,
],
vec![
vec!["restart_count", "tick_count", "price"],
vec!["temperature", "description", "status"],
],
vec![
vec![Value::Int32(1), Value::Int64(2018), Value::Double(1988.1)],
vec![
Value::Float(36.8),
Value::Text("thermograph".to_string()),
Value::Bool(false),
],
],
false,
)?;
let mut ts = Local::now().timestamp_millis();
let mut tablet1 = create_tablet(5, ts);
tablet1.sort();
ts += 5;
let mut tablet2 = create_tablet(10, ts);
ts += 10;
let mut tablet3 = create_tablet(2, ts);
tablet1.sort();
session.insert_tablet(&tablet1)?;
tablet2.sort();
tablet3.sort();
session.insert_tablets(vec![&tablet2, &tablet3])?;
session.insert_records_of_one_device(
"root.sg_rs.dev1",
vec![1, 16],
vec![vec!["status"], vec!["status"]],
vec![vec![Value::Bool(true)], vec![Value::Bool(true)]],
true,
)?;
session.delete_data(vec!["root.sg_rs.dev1.status"], 1, 16)?;
let dataset = session.execute_query_statement("select * from root.sg_rs.device2", None)?;
let mut table = Table::new();
table.set_titles(Row::new(
dataset
.get_column_names()
.iter()
.map(|c| cell!(c))
.collect(),
));
dataset.for_each(|r: RowRecord| {
table.add_row(Row::new(
r.values.iter().map(|v: &Value| cell!(v)).collect(),
));
});
table.printstd();
let dataset = session.execute_statement("show timeseries", None)?;
let mut table = Table::new();
table.set_titles(Row::new(
dataset
.get_column_names()
.iter()
.map(|c| cell!(c))
.collect(),
));
dataset.for_each(|r: RowRecord| {
table.add_row(Row::new(
r.values.iter().map(|v: &Value| cell!(v)).collect(),
));
});
table.printstd();
session.execute_batch_statement(vec![
"insert into root.sg_rs.dev6(time,s5) values(1,true)",
"insert into root.sg_rs.dev6(time,s5) values(2,true)",
"insert into root.sg_rs.dev6(time,s5) values(3,true)",
])?;
let dataset = session.execute_raw_data_query(
vec![
"root.sg_rs.device2.restart_count",
"root.sg_rs.device2.tick_count",
"root.sg_rs.device2.description",
],
0,
i64::MAX,
)?;
let mut table = Table::new();
table.set_titles(Row::new(
dataset
.get_column_names()
.iter()
.map(|c| cell!(c))
.collect(),
));
dataset.for_each(|r: RowRecord| {
table.add_row(Row::new(
r.values.iter().map(|v: &Value| cell!(v)).collect(),
));
});
table.printstd();
if let Some(dataset) =
session.execute_update_statement("delete timeseries root.sg_rs.dev1.*")?
{
dataset.for_each(|r| println!("timestamp: {} {:?}", r.timestamp, r.values));
}
session.close()?;
Ok(())
}
fn create_tablet(row_count: i32, start_timestamp: i64) -> Tablet {
let mut tablet = Tablet::new(
"root.sg_rs.device2",
vec![
MeasurementSchema::new(
String::from("status"),
TSDataType::Boolean,
TSEncoding::Plain,
TSCompressionType::SNAPPY,
None,
),
MeasurementSchema::new(
String::from("restart_count"),
TSDataType::Int32,
TSEncoding::RLE,
TSCompressionType::SNAPPY,
None,
),
MeasurementSchema::new(
String::from("tick_count"),
TSDataType::Int64,
TSEncoding::RLE,
TSCompressionType::SNAPPY,
None,
),
MeasurementSchema::new(
String::from("temperature"),
TSDataType::Float,
TSEncoding::Plain,
TSCompressionType::SNAPPY,
None,
),
MeasurementSchema::new(
String::from("price"),
TSDataType::Double,
TSEncoding::Gorilla,
TSCompressionType::SNAPPY,
None,
),
MeasurementSchema::new(
String::from("description"),
TSDataType::Text,
TSEncoding::Plain,
TSCompressionType::SNAPPY,
None,
),
],
);
(0..row_count).for_each(|row| {
let ts = start_timestamp + row as i64;
tablet
.add_row(
vec![
Value::Bool(ts % 2 == 0),
Value::Int32(row),
Value::Int64(row as i64),
Value::Float(row as f32 + 0.1),
Value::Double(row as f64 + 0.2),
Value::Text(format!("ts: {}", ts).to_string()),
],
ts,
)
.unwrap_or_else(|err| eprintln!("Add row failed, reason '{}'", err));
});
tablet
}
下载：https://github.com/manlge/iotdb-client-rs/releases
