时序原生操作

EnerOS 将时序数据作为内核原生数据类型，无需外挂时序数据库即可支撑百万点/秒的写入与毫秒级查询。这是「时序原生」理念的直接体现：电网的电压、电流、功率天然是时序数据，操作系统应当原生理解而非通过外部插件支持。

电力系统是「时间密集型」领域：PMU 数据以 100-120 Hz 采样，故障录波达到 10 kHz 以上，状态估计每 100ms 一次。传统架构依赖外部 InfluxDB / TimescaleDB，引入网络往返与序列化开销。EnerOS 将时序引擎内嵌到内核，与拓扑、潮流共享同一内存空间，写入路径短至 1μs。

架构总览

┌─────────────────────────────────────────────────────┐
│  Agent / 应用 / 数字孪生                            │
└──────────────────┬──────────────────────────────────┘
                   │ write / query / aggregate
┌──────────────────┴──────────────────────────────────┐
│  eneros-timeseries (内核时序引擎)                   │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ Writer   │ │ Reader   │ │ Compactor│            │
│  └──────────┘ └──────────┘ └──────────┘            │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ CQ 引擎  │ │ Retention│ │ Downsamp │            │
│  └──────────┘ └──────────┘ └──────────┘            │
└──────────────────┬──────────────────────────────────┘
                   │ 列式存储 + Gorilla 编码
┌──────────────────┴──────────────────────────────────┐
│  LSM-Tree (per-tenant)                              │
│  MemTable → SSTable → Compaction                    │
└─────────────────────────────────────────────────────┘

数据模型

时序数据由三要素构成：测量点（tag）、时间戳、字段值。

use eneros_timeseries::{Series, Sample, Timestamp, Quality};

let mut series = Series::new("bus_1.voltage");
series.push(Sample::new(ts1, 1.02));
series.push(Sample::new(ts2, 1.025));

// 带质量位的样本
series.push(Sample::with_quality(ts3, 1.03, Quality::Good));
series.push(Sample::with_quality(ts4, 0.98, Quality::Suspect));

Sample 字段说明

Quality 枚举

写入

支持单点、批量与流式写入：

use eneros_timeseries::TimeSeriesStore;

let store = TimeSeriesStore::open("/var/eneros/tsdb")?;

// 单点写入
store.write("bus_1.voltage", ts, 1.02).await?;

// 带质量位写入
store.write_with_quality("bus_1.voltage", ts, 1.02, Quality::Good).await?;

// 批量写入（推荐，吞吐高 5x）
let batch = vec![
    Sample::new(ts1, 1.02),
    Sample::new(ts2, 1.025),
    Sample::new(ts3, 1.03),
];
store.write_batch("bus_1.voltage", batch).await?;

// 多测点批量写入
let multi_batch = vec![
    ("bus_1.voltage", Sample::new(ts, 1.02)),
    ("bus_1.frequency", Sample::new(ts, 50.01)),
    ("bus_1.load", Sample::new(ts, 50.0)),
];
store.write_multi(multi_batch).await?;

// 流式写入（用于实时采集）
let mut stream = store.stream("bus_1.voltage");
stream.send(Sample::now(1.02)).await?;
stream.flush().await?;

写入 API 一览

查询

use eneros_timeseries::{Query, Range, Aggregation, Filter};

let result = store.query(
    Query::select("bus_1.voltage")
        .range(Range::last(Duration::hours(1)))
        .downsample(Duration::seconds(1), Aggregation::Avg)
        .filter(Filter::quality(Quality::Good))
).await?;

for sample in result {
    println!("{}: {:.4}", sample.timestamp, sample.value);
}

多测点查询

let multi = store.query(
    Query::select_multi(&["bus_1.voltage", "bus_1.frequency", "bus_1.load"])
        .range(Range::between(t_start, t_end))
        .align(AlignStrategy::Interpolate)
).await?;

for (tag, samples) in multi {
    println!("测点 {}: {} 条样本", tag, samples.len());
}

Range 类型

Filter 类型

聚合与降采样

内置常用聚合函数，并支持滚动窗口与滑动窗口：

// 整体聚合
let avg = store.aggregate(
    "bus_1.voltage",
    Range::last(Duration::hours(24)),
    Aggregation::Avg,
).await?;

// 滚动窗口聚合
let windowed = store.window_aggregate(
    "bus_1.load",
    Range::last(Duration::hours(24)),
    Duration::hours(1),
    Aggregation::Percentile(95),
).await?;

for window in &windowed {
    println!("{} ~ {}: P95 = {:.2}", window.start, window.end, window.value);
}

// 滑动窗口聚合
let sliding = store.sliding_window(
    "bus_1.load",
    Range::last(Duration::hours(1)),
    Duration::minutes(15),  // 窗口大小
    Duration::minutes(5),   // 步长
    Aggregation::StdDev,
).await?;

连续查询

注册自动执行的连续查询，避免重复计算：

use eneros_timeseries::ContinuousQuery;

// SQL 风格 CQ
store.register_cq(
    "voltage_5min_avg",
    "SELECT avg(value) INTO bus_1.voltage_5m FROM bus_1.voltage GROUP BY time(5m)",
).await?;

// 类型化 CQ
let cq = ContinuousQuery::new("load_p95_hourly")
    .source("bus_1.load")
    .into("bus_1.load_p95_1h")
    .aggregate(Aggregation::Percentile(95))
    .window(Duration::hours(1))
    .build();
store.register_cq_typed(cq).await?;

// 列出所有 CQ
let cqs = store.list_cqs().await?;
for cq in &cqs {
    println!("{}: {} (last_run: {:?})", cq.name, cq.sql, cq.last_run);
}

// 删除 CQ
store.drop_cq("voltage_5min_avg").await?;

数据保留策略

支持按测量点配置保留期与精度降级：

use eneros_timeseries::RetentionPolicy;

store.set_retention("bus_1.voltage",
    RetentionPolicy::new()
        .raw_for(Duration::days(7))                              // 原始数据保留 7 天
        .downsampled(Duration::seconds(1), Duration::days(90))   // 1s 粒度保留 90 天
        .downsampled(Duration::minutes(1), Duration::years(5))   // 1min 粒度保留 5 年
        .downsampled(Duration::hours(1), Duration::years(20))    // 1h 粒度保留 20 年
).await?;

// 批量配置
let policy = RetentionPolicy::default()
    .raw_for(Duration::days(30));
store.set_retention_multi(&["bus_1.*", "bus_2.*"], policy).await?;

RetentionPolicy 层级

异常检测

内置时序异常检测算法，触发事件通知 Agent：

use eneros_timeseries::{AnomalyDetector, AnomalyKind};

let detector = AnomalyDetector::new("bus_1.voltage")
    .method(AnomalyKind::Spike { threshold: 3.0 })  // 3σ 突变
    .method(AnomalyKind::LevelShift { window: 60 })  // 60s 均值漂移
    .method(AnomalyKind::Flatline { window: 30 })    // 30s 平直
    .on_anomaly(|event| {
        log::warn!("异常: {:?} at {}", event.kind, event.timestamp);
    });

store.register_detector(detector).await?;

存储引擎

底层采用列式存储 + Gorilla 编码 + Delta-of-Delta 时间戳压缩，单节点可承载 10 年电网历史数据。

压缩算法

LSM-Tree 结构

MemTable (内存)
   │ flush
   ▼
SSTable L0 (磁盘，10MB)
   │ compaction
   ▼
SSTable L1 (磁盘，100MB)
   │ compaction
   ▼
SSTable L2+ (磁盘，1GB+)

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04，NVMe SSD。

配置参数

eneros.toml 中时序相关配置：

[timeseries]
# 存储路径
data_dir = "/var/eneros/tsdb"
# MemTable 大小（MB）
memtable_mb = 64
# SSTable 压缩算法：snappy / lz4 / zstd
compression = "lz4"
# Compaction 线程数
compaction_threads = 2
# 默认保留期（天）
default_retention_days = 365
# 是否启用连续查询
enable_cq = true
# CQ 执行间隔（秒）
cq_interval_secs = 10
# 是否启用异常检测
enable_anomaly = true
# 写入缓存大小（MB）
write_buffer_mb = 128

与其他能力的关系

相关文档

• 数字孪生引擎 — 时序数据驱动孪生回放
• 物联网泛在接入 — IoT 设备时序采集
• 电网分析进阶 — 状态估计依赖时序数据
• 安全守卫 — WORM 审计日志
• EnerOS 简介 — Time-Series Native 设计哲学