Fluent Bit + SQL

您可以在这里中找到 BNF 形式的详细查询语言语法。下一节将简要介绍如何使用 SQL 查询语句进行 Fluent Bit 流处理。

Synopsis

Description

从来自数据流或与特定模式匹配的记录中选择关键字。请注意，与流创建不关联的简单的语句会将结果发送到标准输出接口(stdout)，这对于调试目的来说很有用

查询语句支持使用 WHERE 语句来过滤结果。我们将在后面的 aggregation 聚合函数部分解释 WINDOW 和 GROUP BY 语句。

Examples

从来自名为 apache 的流记录中选择所有键:

SELECT * FROM STREAM:apache;

从以 apache. 开头的标签的记录中选择所有键:

SELECT code AS http_status FROM TAG:'apache.*';

CREATE STREAM Statement

Synopsis

CREATE STREAM stream_name
  [WITH (property_name=value, [...])]
  AS select_statement

Description

使用 SELECT 语句的结果创建一个新的数据流。如果在 WITH 语句中设置了属性，则可以选择将新创建的数据流重新输入到 Fluent Bit 管道中

Examples

CREATE STREAM hello AS SELECT * FROM STREAM:apache;

为所有原始标签以 apache 开头的记录创建一个名为 hello 的新数据流:

CREATE STREAM hello AS SELECT * FROM TAG:'apache.*';

聚合函数在选择关键字的 results_statement(结果声明)中使用，从而可以对记录进行分组数据计算。聚合函数应用的记录组由 WINDOW 关键字确定。如果指定 WINDOW 关键字，则聚合函数将应用在接收到记录的当前缓冲区，该记录可能具有不确定的元素数量。可以将聚合函数应用于特定时间间隔窗口中的记录(请参见 select 语句中 WINDOW 的语法)。

Fluent Bit 支持窗口滚动，这是非重叠窗口类型。这意味着大小为 5 秒的窗口将对 5 秒间隔内的记录进行聚合运算，然后为下一时间间隔开始新的聚合运算。

另外，该语法支持 GROUP BY 语句，当 GROUP BY 指定的键具有相同的值时，结果按照一个或多个键进行分组。

AVG

Synopsis

Description

计算 POST 请求的平均值大小

Synopsis

SELECT host, COUNT(*) FROM STREAM:apache WINDOW TUMBLING (5 SECOND) GROUP BY host;

Description

按主机 IP 地址计算 5 秒滑动窗口中的记录数

MIN

Synopsis

SELECT MIN(key) FROM STREAM:apache;

Description

获取一组记录中指定键的值的最小值

MAX

Synopsis

Description

Synopsis

SELECT SUM(key) FROM STREAM:apache;

Description

计算一组记录中键的所有值的总和

时间函数添加新键到带有计时数据的记录中

NOW

Synopsis

SELECT NOW() FROM STREAM:apache;

Description

使用如下格式添加当前系统时间: %Y-%m-%d %H:%M:%S. 输出示例: 2019-03-09 21:36:05.

UNIX_TIMESTAMP

Synopsis

Description

将当前的 Unix 时间戳添加到记录中. 输出示例: 1552196165 .

Record Functions 记录函数使用记录上下文中的值将新键添加到记录

Synopsis

SELECT RECORD_TAG() FROM STREAM:apache;

Description

将记录的标签(字符串)追加为记录的新关键字

RECORD_TIME

Synopsis

SELECT RECORD_TIME() FROM STREAM:apache;

与常规 SQL 语句类似，Fluent Bit 查询语言支持 WHERE 条件。该语言支持对键和子键的进行条件筛选，例如：

SELECT AVG(size) FROM STREAM:apache WHERE method = 'POST' AND status = 200;

SELECT MAX(key) FROM STREAM:apache WHERE @record.contains(key);

检查键的值是否为NULL:

SELECT MAX(key) FROM STREAM:apache WHERE key IS NULL;