在处理时序数据时，数据写入周期通常与数据采集设备的频率相关，有时每秒钟就需要处理大量的数据点。长时间处理如此多的数据会导致存储问题。一个有效的解决方案是使用流式计算，将原始数据进行降采样。

流式计算在时序数据库中指对实时数据流进行计算和处理，以便在数据到达时立即得到计算结果。通过流式计算，我们可以实时地对数据进行处理，例如降采样、聚合、过滤等操作，从而减少数据存储量，并获得实时的统计信息。

流式计算的基本原理是通过对实时数据流进行处理，生成所需的计算结果。例如，将每分钟的数据点合并为每小时的数据点。合并过程可以使用各种统计方法，如最大值、平均值、总和等。

CnosDB 提供了 STREAM 类型的表，用于处理流式数据计算。流查询要求 SELECT 函数中必须包含 GROUP BY time()子句。通过流查询，我们可以实时地对数据进行处理，并将结果存储到目标表中。这在其他数据库中的概念类似物化视图。

创建 CnosDB Stream 表的语法为：

需要一个 source 表，STREAM 表不支持 ALTER

CREATE STREAM TABLE [IF NOT EXISTS] table_name[(field_definition [, field_definition] ...)]    WITH (db = 'db_name', table = 'table_name', event_time_column = 'time_column')    engine = tskv;
field_definition:     column_name data_type
复制

以下是一个示意图，展示了流式计算中的降采样过程：

实现步骤

前置条件：

需要有一个用于流计算的原始表，通过流表视图可以实时收到原表中写入的数据，在流表中建立的流计算定时任务会定时执行，处理原表中的数据，并将其写入目标表。

以下示例使用 CnosDB 中 快速开始 中的示例数据：

\w oceanic_station.txt
复制

1. 创建流表视图

首先，需要创建一个 流表视图 air_stream，该视图从原始表 air 中读取数据，并在流查询中使用。

CREATE STREAM TABLE air_stream(    time TIMESTAMP,    station STRING,    pressure DOUBLE,    temperature DOUBLE,    visibility DOUBLE) WITH (    db = 'oceanic_station',     table = 'air',     event_time_column = 'time')engine = tskv;
复制

2. 创建目标表

然后，创建一个目标表 air_down_sampling_1hour，用于存储降采样后的数据。

CREATE TABLE air_down_sampling_1hour(    max_pressure DOUBLE,     avg_temperature DOUBLE,     sum_temperature DOUBLE,     count_pressure BIGINT,     TAGS(station));
复制

3. 执行流查询并将结果插入目标表

通过流查询将降采样后的数据插入到目标表中。该查询每小时计算一次压力的最大值、温度的平均值、温度的总和以及数据行数。

INSERT INTO air_down_sampling_1hour(    time,     station,     max_pressure,     avg_temperature,     sum_temperature,     count_pressure) SELECT     date_bin(INTERVAL '1' HOUR, time, TIMESTAMP '2023-01-14T16:00:00') time,     station,     MAX(pressure) max_pressure,     AVG(temperature) avg_temperature,     SUM(temperature) sum_temperature,     COUNT(pressure) count_pressure FROM air_stream GROUP BY date_bin(INTERVAL '1' HOUR, time, TIMESTAMP '2023-01-14T16:00:00'), station;
复制

4. 查询降采样结果

当原表中有数据写入后，通过流表视图中任务可以实时计算，并写入目标表 air_down_sampling_1hour中，查看降采样后的数据。

SELECT * FROM air_down_sampling_1hour LIMIT 10;
复制

以上查询将收到以下结果：