用例
持续聚合的三个主要用例示例如下:
-
实时分析:一个实时分析平台,不断聚合来自事件流的数据,提供即时洞察,同时可选择将数据降采样到较低分辨率。例如,此系统可以编译来自高频日志事件流(例如,每毫秒发生一次)的数据,以提供每分钟的请求数、平均响应时间和每分钟的错误率等最新洞察。
-
实时监控:一个实时监控系统,不断聚合来自事件流的数据,根据聚合数据提供实时警报。例如,此系统可以处理来自传感器事件流的数据,以提供当温度超过某个阈值时的实时警报。
-
实时仪表盘:一个实时仪表盘,显示每分钟的请求数、平均响应时间和每分钟的错误数。此仪表板可用于监控系统的健康状况,并检测系统中的任何异常。
在所有这些用例中,持续聚合系统不断聚合来自事件流的数据,并根据聚合数据提供实时洞察和警报。系统还可以将数据降采样到较低分辨率,以减少存储和处理的数据量。这使得系统能够提供实时洞察和警报,同时保持较低的数据存储和处理成本。
实时分析示例
请参阅概述中的实时分析示例。该示例用于计算日志的总数、包大小的最小、最大和平均值,以及大小大于 550 的数据包数量按照每个状态码在 1 分钟固定窗口中的实时分析。
另外,您还可以使用持续聚合来计算其他类型的实时分析。例如,要从 ngx_access_log
表中获取所有不同的国家。持续聚合的查询如下:
/* input table */
CREATE TABLE ngx_access_log (
client STRING,
country STRING,
access_time TIMESTAMP TIME INDEX
);
/* sink table */
CREATE TABLE ngx_country (
country STRING,
update_at TIMESTAMP,
__ts_placeholder TIMESTAMP TIME INDEX,
PRIMARY KEY(country)
);
/* create flow task to calculate the distinct country */
CREATE FLOW calc_ngx_country
SINK TO ngx_country
AS
SELECT
DISTINCT country,
FROM ngx_access_log;
创建好 flow 任务后,我们可以将一些数据插入源表 ngx_access_log
中:
/* insert some data */
INSERT INTO ngx_access_log VALUES
("client1", "US", "2022-01-01 00:00:00"),
("client2", "US", "2022-01-01 00:00:01"),
("client3", "UK", "2022-01-01 00:00:02"),
("client4", "UK", "2022-01-01 00:00:03"),
("client5", "CN", "2022-01-01 00:00:04"),
("client6", "CN", "2022-01-01 00:00:05"),
("client7", "JP", "2022-01-01 00:00:06"),
("client8", "JP", "2022-01-01 00:00:07"),
("client9", "KR", "2022-01-01 00:00:08"),
("client10", "KR", "2022-01-01 00:00:09");
等待一秒钟确保 Flow 有时间将结果写入 sink 表,然后就可以查询结果了:
/* check the result */
select * from ngx_country;
或者,如果您想要按时间窗口对数据进行分组,可以使用以下查询:
/* input table create same as above */
/* sink table */
CREATE TABLE ngx_country (
country STRING,
time_window TIMESTAMP TIME INDEX,
update_at TIMESTAMP,
PRIMARY KEY(country)
);
CREATE FLOW calc_ngx_country
SINK TO ngx_country
AS
SELECT
DISTINCT country,
date_bin(INTERVAL '1 hour', access_time) as time_window,
FROM ngx_access_log
GROUP BY
country,
time_window;
/* insert data using the same data as above */
上述的查询将 ngx_access_log
表中的数据放入 ngx_country
表中。它计算每个时间窗口的不同国家。date_bin
函数用于将数据分组为一小时的间隔。ngx_country
表将不断更新聚合数据,提供实时洞察,显示正在访问系统的不同国家。
请注意,目前 Flow 的内部状态没有持久存储。内部状态指的是用于计算增量查询结果的中间状态,例如聚合查询的累加器值(如count(col)的累加器记录了目前为止的 count 计数)。然而,Sink 表的数据是有持久存储的。因此,建议您使用适当的时间窗口(例如设置为每小时)来最小化数据丢失。因为一旦内部状态丢失,相关时间窗口的数据也将随之丢失。