流计算概述

什么是流计算

流计算用于持续接收外部事件流，并在数据到达时完成处理和写入。相比先落库再离线处理的方式，它更适合实时监控、告警预处理和在线数据清洗等低延迟场景。

在 Datalayers 中，流计算链路由三部分组成：

• SOURCE：定义外部数据流的 schema、connector 和 format
• PIPELINE：定义实时处理逻辑
• SINK TABLE：接收 pipeline 输出结果的内部表，目前必须是 TimeSeries 表

流程如下：

Kafka / MQTT / HTTP
        |
        v
   CREATE SOURCE
        |
        v
  CREATE PIPELINE ... AS SELECT ...
        |
        v
   TimeSeries sink table

应用场景

• 从 Kafka、MQTT、HTTP 持续接收或拉取数据
• 在数据入库前做字段筛选、投影、阈值过滤等实时计算
• 将清洗、转换后的结果写入内部时序表，供 SQL 查询、看板或告警系统使用

当前 PIPELINE 仅支持基于单个 source 的投影和过滤，暂不支持 join、聚合、窗口、排序、limit、union、子查询等复杂算子。

核心 SQL 对象

Source

SOURCE 用于声明输入流的列定义、connector 和 format。它本身不保存数据，只负责将外部消息解码为表结构。示例：

CREATE SOURCE src_kafka (
  ts TIMESTAMP(9) NOT NULL,
  sid STRING NOT NULL,
  value FLOAT64
) WITH (
  connector='kafka',
  brokers='127.0.0.1:9092',
  topic='topic_stream_demo',
  format='json'
);

Pipeline

PIPELINE 表示持续运行的实时任务，负责从 source 读取数据、执行处理并写入 sink table。

CREATE PIPELINE p_kafka
SINK TO sink_t
AS
SELECT ts, sid, value
FROM src_kafka
WHERE value >= 2.0;

Sink Table

SINK 不是独立对象，而是一张已存在的内部表。当前必须使用 TimeSeries 引擎，且查询输出 schema 必须与 sink table 严格兼容。

Connectors 与 Formats

Connector 用于描述外部数据系统，决定 source 如何读取数据。

Format 用于描述消息编码格式，Datalayers 会据此将原始消息解码为表结构。

当前版本支持的 connectors 请参考 Connectors。

当前版本支持的 formats 请参考 Formats。

管理与运维

流计算对象创建后，可以用以下语句查看和控制：

# 查看当前数据库下的所有 source
SHOW SOURCES;

# 查看当前数据库下的所有 pipeline
SHOW PIPELINES;

# 查看指定 source 的定义 SQL
SHOW CREATE SOURCE src_kafka;

# 查看指定 pipeline 的定义 SQL
SHOW CREATE PIPELINE p_kafka;

# 停止一个运行中的 pipeline
ALTER PIPELINE p_kafka STOP;

# 重启一个 pipeline
ALTER PIPELINE p_kafka RESTART;

# 删除指定 pipeline
DROP PIPELINE p_kafka;

# 删除指定 source
DROP SOURCE src_kafka;

相关文档

• 了解如何快速体验一条最小流计算链路：快速开始
• 了解 Datalayers 的流计算模型 ：计算模型
• 了解流计算的典型业务场景和示例：应用示例
• 了解 JSON、CSV 等消息格式及其配置方式：Formats
• 了解当前支持的 connector 类型和配置方式：Connectors