#influx-db #reqwest-client #time-series #client-send #data-query

rinfluxdb-lineprotocol

用于查询和向InfluxDB发送数据的库

1 个不稳定版本

0.2.0 2021年11月27日

#2209数据库接口

Download history 9/week @ 2024-03-12 2/week @ 2024-03-26 20/week @ 2024-04-02 2/week @ 2024-04-23 1/week @ 2024-05-07 4/week @ 2024-05-14 7/week @ 2024-05-21 11/week @ 2024-05-28 27/week @ 2024-06-04 2/week @ 2024-06-11 1/week @ 2024-06-18 23/week @ 2024-06-25

57 每月下载次数
用于 rinfluxdb

MIT/Apache

45KB
674

Rust InfluxDB 库

用于查询和向InfluxDB发送数据的库。

https://gitlab.com/claudiomattera/rinfluxdb

特性

  • 将数据序列化为InfluxDB行协议;
  • 在Rust中构建InfluxQL查询;
  • 在Rust中构建FLUX查询;
  • 解析InfluxDB的响应;
  • (可选) 基于Reqwest的客户端以执行常见查询。
  • 解析InfluxQL查询返回的InfluxDB JSON中的数据框;
  • 解析FLUX查询返回的InfluxDB注解CSV中的数据框;
  • (可选) Reqwest对象的包装器,用于构建请求和解析响应;

将数据序列化为InfluxDB行协议

行协议用于向InfluxDB发送数据。

use rinfluxdb::line_protocol::LineBuilder;
use chrono::{TimeZone, Utc};

let line = LineBuilder::new("location")
    .insert_field("latitude", 55.383333)
    .insert_field("longitude", 10.383333)
    .insert_tag("city", "Odense")
    .set_timestamp(Utc.ymd(2014, 7, 8).and_hms(9, 10, 11))
    .build();

assert_eq!(line.measurement(), &"location".into());
assert_eq!(line.field("latitude"), Some(&55.383333.into()));
assert_eq!(line.field("longitude"), Some(&10.383333.into()));
assert_eq!(line.tag("city"), Some(&"Odense".into()));
assert_eq!(line.timestamp(), Some(&Utc.ymd(2014, 7, 8).and_hms(9, 10, 11)));

assert_eq!(
    line.to_string(), 
    "location,city=Odense latitude=55.383333,longitude=10.383333 1404810611000000000"
);

在Rust中构建InfluxQL查询

可以使用 influxql::QueryBuilder 构建InfluxQL查询。

use rinfluxdb::influxql::QueryBuilder;
use chrono::{TimeZone, Utc};

let query = QueryBuilder::from("indoor_environment")
    .field("temperature")
    .field("humidity")
    .start(Utc.ymd(2021, 3, 7).and_hms(21, 0, 0))
    .build();

assert_eq!(
    query.as_ref(),
    "SELECT temperature, humidity \
    FROM indoor_environment \
    WHERE time > '2021-03-07T21:00:00Z'",
);

在Rust中构建FLUX查询

可以使用 flux::QueryBuilder 构建FLUX查询。

use rinfluxdb::types::Duration;
use rinfluxdb::flux::QueryBuilder;

let query = QueryBuilder::from("telegraf/autogen")
    .range_start(Duration::Minutes(-15))
    .filter(
        r#"r._measurement == "cpu" and
        r._field == "usage_system" and
        r.cpu == "cpu-total""#
    )
    .build();

assert_eq!(
    query.as_ref(),
    r#"from(bucket: "telegraf/autogen")
  |> range(start: -15m)
  |> filter(fn: (r) =>
    r._measurement == "cpu" and
    r._field == "usage_system" and
    r.cpu == "cpu-total"
  )
  |> yield()"#,
);

解析InfluxDB的响应

向InfluxDB发送查询时,它会以JSON或注解CSV内容的形式回复,其中包含数据框列表。此库允许将此类回复解析为用户定义的数据框类型。

数据框必须可以从其名称(一个字符串)、其索引(一个瞬间向量)及其列(列名称到值向量的映射)构建。

数据框实现只需要实现此特性即可与该crate一起使用。即,只要为给定的类型 DF (和类型 E 实现 TryFrom<(String, Vec<DateTime<Utc>>, HashMap<String, Vec<Value>>), Error = E>) 实现,解析器就可以使用它来构建最终对象。

一个DataFrame的示例实现可用作 dataframe::DataFrame,但此特性能应用于许多其他现有库。

解析InfluxQL查询返回的JSON

InfluxQL查询的JSON响应可以被解析为DataFrame。

use rinfluxdb::influxql::{ResponseError, StatementResult, from_str};
use rinfluxdb::dataframe::DataFrame;

let input: String = todo!();

let response: Result<Vec<StatementResult<DataFrame>>, ResponseError> =
    from_str(&input);

解析FLUX查询返回的注释CSV

FLUX查询的注释CSV响应可以被解析。

use rinfluxdb::flux::{ResponseError, from_str};
use rinfluxdb::dataframe::DataFrame;

let input: String = todo!();

let response: Result<DataFrame, ResponseError> = from_str(&input);

(可选) 使用基于Reqwest的客户端执行常见查询

上述函数可用于序列化和反序列化查询和数据到原始文本,并且它们可以被集成到现有应用中。作为替代,此库还实现了一个基于Reqwest的可选客户端API,以直接与InfluxDB实例交互。提供阻塞和非阻塞客户端,并支持常见查询。

使用client Cargo功能启用客户端。

使用InfluxQL查询InfluxDB

# use std::collections::HashMap;
# use url::Url;
#
use rinfluxdb::influxql::QueryBuilder;
use rinfluxdb::influxql::blocking::Client;
use rinfluxdb::dataframe::DataFrame;

let client = Client::new(
    Url::parse("https://example.com/")?,
    Some(("username", "password")),
)?;

let query = QueryBuilder::from("indoor_environment")
    .database("house")
    .field("temperature")
    .field("humidity")
    .build();
let dataframe: DataFrame = client.fetch_dataframe(query)?;
println!("{}", dataframe);

let query = QueryBuilder::from("indoor_environment")
    .database("house")
    .field("temperature")
    .field("humidity")
    .group_by("room")
    .build();
let tagged_dataframes: HashMap<String, DataFrame> = 
    client.fetch_dataframes_by_tag(query, "room")?;
for (tag, dataframe) in tagged_dataframes {
    println!("{}: {}", tag, dataframe);
}

# Ok::<(), rinfluxdb::influxql::ClientError>(())

使用FLUX查询InfluxDB

unimplemented!()

使用行协议将数据发送到InfluxDB

# use url::Url;
#
use rinfluxdb::line_protocol::LineBuilder;
use rinfluxdb::line_protocol::blocking::Client;

let client = Client::new(
    Url::parse("https://example.com/")?,
    Some(("username", "password")),
)?;

let lines = vec![
    LineBuilder::new("measurement")
        .insert_field("field", 42.0)
        .build(),
    LineBuilder::new("measurement")
        .insert_field("field", 43.0)
        .insert_tag("tag", "value")
        .build(),
];

client.send("database", &lines)?;

# Ok::<(), rinfluxdb::line_protocol::ClientError>(())

(可选) Reqwest对象包装器以构造请求和解析响应

此crate通过HTTP(s)与InfluxDB实例通信。HTTP请求和响应的内容必须遵循InfluxDB约定和协议,但它们可以被定制,例如通过添加基本身份验证。

为了确保最大的自由度,构建了一个围绕Reqwest的Client的微小包装器,使其能够创建一个已准备好与InfluxDB通信的请求构建器。然后可以将此构建器转换为常规请求构建器并执行。

# use url::Url;
#
use rinfluxdb::influxql::Query;

// Bring into scope the trait implementation
use rinfluxdb::influxql::blocking::InfluxqlClientWrapper;

// Create Reqwest client
let client = reqwest::blocking::Client::new();

// Create InfluxQL request
let base_url = Url::parse("https://example.com")?;
let mut builder = client
    // (this is a function added by the trait above)
    .influxql(&base_url)?
    // (this functions are defined on influxql::RequestBuilder)
    .database("house")
    .query(Query::new("SELECT temperature FROM indoor_temperature"))
    // (this function returns a regular Reqwest builder)
    .into_reqwest_builder();

// Now this is a regular Reqwest builder, and can be customized as usual
if let Some((username, password)) = Some(("username", "password")) {
    builder = builder.basic_auth(username, Some(password));
}

// Create a request from the builder
let request = builder.build()?;

// Execute the request through Reqwest and obtain a response
let response = client.execute(request)?;

# Ok::<(), rinfluxdb::influxql::ClientError>(())

同样,也构建了一个围绕Reqwest的Response的微小包装器,以便添加一个新功能来从其中解析DataFrame。

# use std::collections::HashMap;
#
# use url::Url;
#
use rinfluxdb::influxql::Query;

use rinfluxdb::influxql::StatementResult;
use rinfluxdb::influxql::blocking::InfluxqlClientWrapper;
use rinfluxdb::dataframe::DataFrame;

// Bring into scope the trait implementation
use rinfluxdb::influxql::blocking::InfluxqlResponseWrapper;

// Create Reqwest client
let client = reqwest::blocking::Client::new();

// Create InfluxQL request
let base_url = Url::parse("https://example.com")?;
let mut request = client
    .influxql(&base_url)?
    .database("house")
    .query(Query::new("SELECT temperature FROM indoor_temperature"))
    .into_reqwest_builder()
    .build()?;

// Execute the request through Reqwest and obtain a response
let response = client.execute(request)?;

// Return an error if response status is not 200
// (this is a function from Reqwest's response)
let response = response.error_for_status()?;

// Parse the response from JSON to a list of dataframes
// (this is a function added by the trait above)
let results: Vec<StatementResult<DataFrame>> = response.dataframes()?;

# Ok::<(), rinfluxdb::influxql::ClientError>(())

为Reqwest的阻塞API(influxql::blocking::InfluxqlClientWrapperinfluxql::blocking::InfluxqlResponseWrapper)和异步API(influxql::r#async::InfluxqlClientWrapperinfluxql::r#async::InfluxqlResponseWrapper)定义包装器,并使用client Cargo功能启用。

用法

此crate是较小的crate的简单聚合器,每个crate都通过Cargo功能启用并实现InfluxDB支持的一个特定部分。

rinfluxdb
├── rinfluxdb-types
├── rinfluxdb-lineprotocol
├── rinfluxdb-influxql
├── rinfluxdb-flux
└── rinfluxdb-dataframe

客户端可以依赖rinfluxdb启用必要的功能,或者它们可以显式依赖rinfluxdb-* crates。

[dependencies.rinfluxdb]
version = "0.2.0"
features = ["lineprotocol", "influxql", "client"]

# Or

[dependencies]
rinfluxdb-lineprotocol = { version = "0.2.0", features = ["client"] }
rinfluxdb-influxql = { version = "0.2.0", features = ["client"] }

Cargo功能

此crate支持以下Cargo功能。

  • lineprotocol:重新导出rinfluxdb-lineprotocol crate;
  • influxql:重新导出rinfluxdb-influxql crate;
  • flux:重新导出rinfluxdb-flux crate;
  • dataframe:重新导出rinfluxdb-dataframe crate;
  • client:在所有rinfluxdb-* crates中启用client功能。

当启用功能 client 时,crate 为行协议、InfluxQL 和 Flux 定义客户端。客户端使用 Reqwest 实现,并提供阻塞和异步模式。

许可证

版权所有 Claudio Mattera 2021

您可以在以下任一许可证的条款下免费复制、修改和分发此应用程序:

任选其一。

本项目完全是原创作品,与 InfluxData 没有任何关联,也不获得其任何形式的认可。

依赖项

~2–17MB
~192K SLoC