ksql

ksqlDB 流处理命令行

补充说明

ksql 连接 ksqlDB 服务器，用 SQL 创建流、表、物化视图和持续查询。它把 Kafka topic 映射成 STREAM/TABLE，适合实时 ETL 和告警。

常用示例

ksql http://ksqldb:8088
ksql http://ksqldb:8088 --file queries.sql

SHOW STREAMS;
SHOW TABLES;
SHOW QUERIES;

CREATE STREAM app_events (
  id INT,
  event_type VARCHAR,
  ts BIGINT
) WITH (
  KAFKA_TOPIC='app-events',
  VALUE_FORMAT='JSON'
);

CREATE TABLE event_counts AS
SELECT event_type, COUNT(*) AS cnt
FROM app_events
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY event_type
EMIT CHANGES;

SELECT * FROM event_counts EMIT CHANGES;
DESCRIBE app_events;
TERMINATE <query-id>;
DROP STREAM app_events;

易错点与排错

• STREAM 是追加事件，TABLE 是变更日志；选错模型会导致结果不符合预期。
• EMIT CHANGES 是持续查询；普通 SELECT 和 pull query 的语义不同。
• 窗口聚合会产生大量状态；窗口大小和 retention 要按业务设置。
• Schema 和序列化格式必须与 topic 一致；JSON/Avro/Protobuf 配置不同。
• 持久查询会占用资源和状态存储；TERMINATE 后再决定是否删除。
• ksqlDB 集群依赖 Kafka 和 Schema Registry；底层不可用时查询会失败。
• 生产变更要通过迁移脚本管理，避免手工改流表后无法复现。

相关命令

• kafka-broker — 底层 Kafka
• schema-registry — Schema 管理
• kcat — 消息验证