用户指南 | 如何使用 Flow 功能实现持续聚合，赋能实时计算和查询

持续聚合功能可以完全通过 SQL 来定义和使用，本文将演示从创建 Flow 任务，接受数据进行流处理，到删除该 Flow 任务的全部流程。

我们以一个速度传感器读入左右轮的瞬时速度，并且过滤掉较高或较低的异常值，并计算五秒内的平均速度为例。

CREATE TABLE velocity (
    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    left_wheel FLOAT,
    right_wheel FLOAT,
    TIME INDEX(ts)
);

CREATE TABLE avg_speed (
    avg_speed FLOAT,
    start_window TIMESTAMP TIME INDEX,
    end_window TIMESTAMP,
    update_at TIMESTAMP,
);

接下来就可以创建 Flow 任务了，这里需要使用我们提供的 SQL 方言语法 CREATE FLOW，可以参考如下例子：

CREATE FLOW calc_avg_speed
SINK TO avg_speed
AS
SELECT avg((left_wheel+right_wheel)/2)
FROM velocity
WHERE left_wheel > 0.5 AND right_wheel > 0.5 AND left_wheel < 60 AND right_wheel < 60
GROUP BY tumble(ts, '5 second');

上述 SQL 语句的意思是：创建一个名为 calc_avg_speed 的 Flow 任务，它会把结果输出到 avg_speed 表上。其上运行的查询由 AS 之后的 SELECT 语句定义。

首先，过滤掉左右轮速度值中过小或过大的值（小于等于 0.5 或大于等于 60）；然后，基于输入表 velocity，以 ts 列的每五秒的间隔作为窗口，计算窗口内的平均速度。Flow 作业当中的查询完全基于 SQL 语法，并根据需求实现了相关扩展。

INSERT INTO velocity 
VALUES
    ("2021-07-01 00:00:00.200", 0.0, 0.7),
    ("2021-07-01 00:00:00.200", 0.0, 61.0),
    ("2021-07-01 00:00:02.500", 2.0, 1.0,);

SELECT * FROM avg_speed;

 avg_speed |        start_window        |         end_window         |         update_at          
-----------+----------------------------+----------------------------+----------------------------
       1.5 | 2021-07-01 00:00:00.000000 | 2021-07-01 00:00:05.000000 | 2024-06-04 03:35:20.670000
(1 row)

INSERT INTO velocity 
VALUES
    ("2021-07-01 00:00:05.100", 5.0, 4.0),
    ("2021-07-01 00:00:09.600", 2.3, 2.1);

SELECT * FROM avg_speed;

 avg_speed |        start_window        |         end_window         |         update_at          
-----------+----------------------------+----------------------------+----------------------------
       1.5 | 2021-07-01 00:00:00.000000 | 2021-07-01 00:00:05.000000 | 2024-06-04 03:35:20.670000
      3.35 | 2021-07-01 00:00:05.000000 | 2021-07-01 00:00:10.000000 | 2024-06-04 03:35:34.693000

avg_speed 表中的列解释如下：

• avg_speed：窗口中计算得到的平均速度；

• start_window：窗口的开始时间；

• end_window：窗口的结束时间；

• update_at：更新行数据的时间。

最后，使用 DROP FLOW 删除这个 Flow 任务：

DROP FLOW calc_avg_speed;

CREATE FLOW [ IF NOT EXISTS ] <flow-name>
SINK TO <sink-table-name>
[ EXPIRE AFTER <expr> ]
[ COMMENT = "<string>" ]
AS 
<SQL>;

CREATE FLOW IF NOT EXISTS my_flow
SINK TO my_sink_table
EXPIRE AFTER INTERVAL '1 hour'
COMMENT = "My first flow in GreptimeDB"
AS
SELECT count(item) from my_source_table GROUP BY tumble(time_index, '5 minutes');

DROP FLOW [IF EXISTS] <name>

本文介绍了 GreptimeDB 中持续聚合功能的基本用法和特性，并且举例说明了创建、使用和删除 Flow 任务的流程。使用持续聚合可以随时、低延时（秒级/亚秒级）地获取用户关心的信息，同时避免了额外的内存和计算上的开销。

未来，除了支持更多函数之外，我们还会支持流处理中间状态的持久化和诸如 Temporal Filter 等高级功能，更详细的信息可以参考相关的 用户文档^[1] 和 开发指南^[2]。

Reference:

[1] https://docs.greptime.com/user-guide/continuous-aggregation/overview

[2] https://docs.greptime.com/contributor-guide/flownode/overview

---

关于 Greptime

Greptime 格睿科技专注于为物联网（如智慧能源、智能汽车等）及可观测等产生大量时序数据的领域提供实时、高效的数据存储和分析服务，帮助客户挖掘数据的深层价值。目前主要有以下三款产品：

GreptimeDB 是一款用 Rust 语言编写的开源时序数据库，具有云原生、无限水平扩展、高性能和融合分析等特点，帮助企业实时读写、处理和分析时序数据的同时，降低长期存储的成本。我们提供 GreptimeDB 企业版，支持更多企业特性和定制化服务，如有需要欢迎联系我们：15310923206（同微信）。

GreptimeCloud 是一款全托管的云上数据库即服务（DBaaS）解决方案，基于开源时序数据库 GreptimeDB 打造，能够高效支持可观测、物联网、金融等领域的应用。用户可以通过内置的可观测性解决方案 GreptimeAI 全面掌握 LLM 应用的成本、性能、流量和安全等情况。

车云一体解决方案 是一款深入车企实际业务场景的时序数据库解决方案，解决了企业车辆数据呈几何倍数增长后的实际业务痛点。多模态车端数据库结合云端 GreptimeDB 企业版帮助车企极大降低流量、计算和存储成本，并帮助提升数据实时性和业务洞察能力。

GreptimeDB 作为开源项目，欢迎对时序数据库、Rust 语言等内容感兴趣的同学们参与贡献和讨论。第一次参与项目的同学推荐先从带有 good first issue 标签的 issue 入手，期待在开源社群里遇见你！扫码添加小助手备注“技术交流群”立即加入讨论吧~

Star us on GitHub Now: 

https://github.com/GreptimeTeam/greptimedb

官网：https://greptime.cn/

文档：https://docs.greptime.cn/

Twitter: https://twitter.com/Greptime

Slack: https://greptime.com/slack

LinkedIn: https://www.linkedin.com/company/greptime/

点击下方链接🔗关注 GreptimeDB，了解更多技术干货👇

往期精彩文章：

👇 点击下方阅读原文，立即体验 GreptimeDB！