> 了解数据库的开发者应该都知道，在传统的数据处理流程过程中，总是要先去收集数据，然后再将数据存放到数据库中。当根据业务逻辑去数据库中对数据做查询操作的时候，才能得到想要的结果或者进行对应的处理操作。虽然这种方式较为常用，但是得出的结果却差强人意，尤其是在一些实时查询业务中的某些具体业务场景，比如MapReduce方式的离线处理并不能很好地解决问题，这就出现了一种新的数据计算结构--流式计算。作为数据库的初学者，我想通过本文来分享一下自己关于流式计算学习使用的心得体会。

**1、什么是流式计算？**

流式计算其实就是数据在输入录入的时候就已经开始计算来，是边录边算的方式进行，而且计算的运算速度极快，可达到毫秒级别，然后计算完成之后就能实时体现结果，或者进行存储操作。常用于电商、智慧交通、视频直播、实时天气、物联网等需要实时推送结果的实际场景。其具有实时性、突发性、易失性、无限性特点。
流式计算典型范式之一就是：不确定数据的流入速率的事件流流入系统，系统的处理能力必须要和事件流量相匹配，或通过近似算法等方式进行降级操作，通常称为负载分流。在数据库学习或者开发时，有关大数据实时处理常用的就是流式计算，而目前方便使用和学习的关于流式计算的工具引擎有很多，但是既支持时间驱动的流式计算又支持事件驱动的流式计算产品却不多见，更遑论数据库自带流式计算功能了，很多数据架构因为叠加太多的第三方软件变得越发臃肿。
针对这种情况，涛思数据打造的时序数据库（Time Series Database）TDengine能够缓解这种数据焦虑，新发布的3.0版本更是对流式计算引擎进行了全新升级，既支持时间驱动的流式计算又支持事件驱动。为了验证官方的说法，我打算来尝试一下。

**2、TDengine3.0的流式计算功能**

这里还要再强调一下为什么选择TDengine来使用，不仅因为TDengine的流式计算能够支持在多个vnode中分布的超级表聚合，而且其还提供了watermark机制以度量容忍数据乱序来处理乱序数据的写入的情况，且提供了ignore expired配置项来决策乱序数据的处理操作。

首先需要进行 TDengine 3.0的安装和部署，在这里我就不多做介绍了，大家可以进入 TDengine 官网-文档进行查阅，很详细。

**TDengine3.0的流式计算语句使用**

**（1）流式计算的创建**
TDengine3.0的流式计算的创建语句，如下所示：
```cpp
CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
stream_options: {
TRIGGER [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
WATERMARK time
}
```
上面的subquery就是select普通查询语法的子集，如下所示：

```cpp
subquery: SELECT select_list
from_clause
[WHERE condition]
[PARTITION BY tag_list]
[window_clause]
```
下面举一个示例来看看上述关于流式计算的创建的使用，该流式计算按照每一分钟为时间窗口，30秒为前向增量统计直播间的平均人数，并把来自meter表的数据计算结果写入到avg_value表中，不一样partition的数据会分别创建子表并对应写入子表中，具体如下所示：
```cpp
CREATE STREAM avg_value_s INTO avg_value AS
SELECT _wstartts, count(*), avg(online) FROM meter PARTITION BY tbname INTERVAL(1m) SLIDING(30s);
```

**（2）流式计算的删除**
TDengine3.0的流式计算的删除语句，如下所示：
```cpp
DROP STREAM [IF NOT EXISTS] stream_name;
```
上述语句只删除流式计算的任务，并不会删除流式计算写入的数据，示例如下所示：
DROP STREAM avg_value_s; //只删除了avg_value_s的流式计算任务

**（3）流式计算的显示**
TDengine3.0的流式计算的显示语句，如下所示：
```cpp
SHOW STREAMS;
```
上述语句的显示不太详细，如需更加详细，使用下面的语句：
```cpp
SELECT * from performance_schema.perfo_str;
```
经过上面TDengine3.0的流式计算语法的基本使用操作，大致就是流式计算使用的全部内容，非常的简单易用，且简洁明了。

**3、TDengine3.0流式计算的实际使用示例**

学习TDengine3.0过程中，结合上面学到的基础语法，接下来分享一下TDengine3.0流式计算的实际使用示例，这里以视频直播中用户走势数据为例，通过TDengine3.0流式计算将10天未互动的用户数据清理掉，然后以每分钟为窗口整合并计算每个窗口中的用户年龄的最大值，最后将结果输出到对应的数据表里面，具体如下所示：

**（1）创建原始表**
前提是先准备数据，然后完成建数据库、建一个超级表和多个子表，具体操作如下所示：

![](http://img.blog.itpub.net/blog/2023/01/06/9a25206a8c1a8884.jpeg?x-oss-process=style/bb)

**（2）创建流**
直接创建流，具体操作如下所示：
```cpp
create stream current_stream into current_output_stb as select _wstart as start, _wend as end, max(current) as max_current from meter where age <= 50 interval (1m);
```

**（3）给表中写入数据**
给表中写入数据，具体代码如下所示：

![](http://img.blog.itpub.net/blog/2023/01/06/2350dac2119fa877.jpeg?x-oss-process=style/bb)

**（4）查看输出结果**
查看运算的结果，具体代码如下所示：

![](http://img.blog.itpub.net/blog/2023/01/06/b593b43f10000e28.jpeg?x-oss-process=style/bb)

上面这个示例是TDengine3.0流式计算的实际使用示例，验证了TDengine3.0流式计算的功能，体验到了它的流畅度，而且TDengine3.0流式计算的功能使用起来其实不复杂，操作简单，体验很不错，学习成本也很低。

**总结**
通过对TDengine3.0流式计算的学习使用，体验到的不仅是它带来的简洁操作，还有流畅的计算性能，简化了系统的复杂度，更节省了很多时间成本，尤其是比以往的传统方式要简单太多。不过我本次学习的深度还不够，还有很多需要操作实践的地方，TDengine3.0流式计算的功能很强大，需要开发者持续去使用和发现。

ITPub博客

既支持时间驱动又支持事件驱动，TDengine3.0流式计算的学习使用心得