CREATE STREAM
Introduced or updated: v1.2.391
STREAM是企业版功能。 如需获取许可证，请联系 Databend 支持团队。
创建一个流。
语法
CREATE [ OR REPLACE ] STREAM [ IF NOT EXISTS ] [ <database_name>. ]<stream_name>
ON TABLE [ <database_name>. ]<table_name>
[ AT ( { TIMESTAMP => <timestamp> | SNAPSHOT => '<snapshot_id>' | STREAM => <existing_stream_name> } ) ]
[ APPEND_ONLY = true | false ]
[ COMMENT = '<comment>' ]
| 参数 | 描述 |
|---|---|
< database_name > | 流被视为属于特定数据库的对象，类似于表或视图。CREATE STREAM 允许流和关联表之间使用不同的数据库。如果没有明确指定数据库，则当前数据库将作为您创建的流的数据库。 |
| AT | 当使用 AT 后跟 TIMESTAMP => 或 SNAPSHOT => 时，您可以通过时间戳或快照 ID 创建一个包含特定历史时间点之后数据变化的流；当 AT 后跟 STREAM => 时，它允许创建一个与现有流相同的新流，保留相同的数据变化。 |
| APPEND_ONLY | 当设置为 true 时，流以 Append-Only 模式运行；当设置为 false 时，它以 Standard 模式运行。默认为 false。有关流操作模式的更多详细信息，请参阅 How Stream Works。 |
示例
此示例演示了创建一个名为 'order_changes' 的流来监控 'orders' 表中的变化：
-- 创建一个名为 'orders' 的表
CREATE TABLE orders (
order_id INT,
product_name VARCHAR(255),
quantity INT,
order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 为 'orders' 表创建一个名为 'order_changes' 的流
CREATE STREAM order_changes ON TABLE orders;
-- 向 'orders' 表插入订单 1001
INSERT INTO orders (order_id, product_name, quantity) VALUES (1001, 'Product A', 10);
-- 向 'orders' 表插入订单 1002
INSERT INTO orders (order_id, product_name, quantity) VALUES (1002, 'Product B', 20);
-- 从 'order_changes' 流中检索所有记录
SELECT * FROM order_changes;
┌───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ──────┐
│ order_id │ product_name │ quantity │ order_date │ change$action │ change$is_update │ change$row_id │
├─────────────────┼──────────────────┼─────────────────┼────────────────────────────┼───────────────┼──────────────────┼────────────────────────────────────────┤
│ 1002 │ Product B │ 20 │ 2024-03-28 03:24:16.629135 │ INSERT │ false │ acb58bd6bb4243a4bf0832bf570b38c2000000 │
│ 1001 │ Product A │ 10 │ 2024-03-28 03:24:16.539178 │ INSERT │ false │ b93a15e694db4134ab5a23afa8c92b20000000 │
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
以下示例创建了一个名为 'order_changes_copy' 的新流，并使用 AT 参数，包含与 'order_changes' 相同的数据变化：
-- 在 'orders' 表上创建一个名为 'order_changes_copy' 的流，复制 'order_changes' 的数据变化
CREATE STREAM order_changes_copy ON TABLE orders AT (STREAM => order_changes);
-- 从 'order_changes_copy' 流中检索所有记录
SELECT * FROM order_changes_copy;
┌───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ order_id │ product_name │ quantity │ order_date │ change$action │ change$is_update │ change$row_id │
├─────────────────┼──────────────────┼─────────────────┼────────────────────────────┼───────────────┼──────────────────┼────────────────────────────────────────┤
│ 1002 │ Product B │ 20 │ 2024-03-28 03:24:16.629135 │ INSERT │ false │ acb58bd6bb4243a4bf0832bf570b38c2000000 │
│ 1001 │ Product A │ 10 │ 2024-03-28 03:24:16.539178 │ INSERT │ false │ b93a15e694db4134ab5a23afa8c92b20000000 │
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
此示例在 'orders' 表上创建了两个流。每个流都使用 AT 参数分别获取特定快照 ID 或时间戳之后的数据变化。
-- 从 'orders' 表中检索快照和时间戳信息
SELECT snapshot_id, timestamp from FUSE_SNAPSHOT('default','orders');
┌───────────────────────────────────────────────── ──────────────┐
│ snapshot_id │ timestamp │
├──────────────────────────────────┼────────────────────────────┤
│ f7f57c7d07f445a68e4aa53fa2578bbb │ 2024-03-28 03:24:16.633721 │
│ 11b9d81eabc94c7da648908f0ba313a1 │ 2024-03-28 03:24:16.611835 │
└───────────────────────────────────────────────────────────────┘
-- 在 'orders' 表上创建一个名为 'order_changes_after_snapshot' 的流，捕获特定快照之后的数据变化
CREATE STREAM order_changes_after_snapshot ON TABLE orders AT (SNAPSHOT => '11b9d81eabc94c7da648908f0ba313a1');
-- 查询 'order_changes_after_snapshot' 流以查看捕获的特定快照之后的数据变化
SELECT * FROM order_changes_after_snapshot;
┌───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ order_id │ product_name │ quantity │ order_date │ change$action │ change$is_update │ change$row_id │
├─────────────────┼──────────────────┼────────────── ───┼────────────────────────────┼───────────────┼──────────────────┼────────────────────────────────────────┤
│ 1002 │ Product B │ 20 │ 2024-03-28 03:24:16.629135 │ INSERT │ false │ acb58bd6bb4243a4bf0832bf570b38c2000000 │
└────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ─────────────────────┘
-- 在 'orders' 表上创建一个名为 'order_changes_after_timestamp' 的流，捕获特定时间戳之后的数据变化
CREATE STREAM order_changes_after_timestamp ON TABLE orders AT (TIMESTAMP => '2024-03-28 03:24:16.611835'::TIMESTAMP);
-- 查询 'order_changes_after_timestamp' 流以查看捕获的特定时间戳之后的数据变化
SELECT * FROM order_changes_after_timestamp;
┌───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ order_id │ product_name │ quantity │ order_date │ change$action │ change$is_update │ change$row_id │
├─────────────────┼──────────────────┼─────────────────┼────────────────────────────┼───────────────┼──────────────────┼────────────────────────────────────────┤
│ 1002 │ Product B │ 20 │ 2024-03-28 03:24:16.629135 │ INSERT │ false │ acb58bd6bb4243a4bf0832bf570b38c2000000 │
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
