流式数据库是把流处理引擎的能力合并到数据库内核，与数据库SQL引擎、存储引擎融合在一起，完全以数据库的习惯使用流处理引擎，甚至可以和数据库的功能混合使用。比如流对象与表对象联合JOIN。

1. 技术栈简单，稳定性好。
2. 学习和维护成本低，只需掌握数据库的使用和维护即可。
3. 纯SQL操作，使用简单方便，可快速响应业务的复杂多变性。
4. 流数据处理支持数据UPDATE和DELETE、事务ACID以及流对象与表对象联合JOIN。

   
   

流

流对象

PULL和PUSH查询

时间概念

• 事件生成时间

• 事件接入时间

• 事件处理时间

窗口

1. 滚动窗口

2. 滑动窗口

3. 会话窗口

4. 全局窗口

水位线

• 条件T1：水位线时间>=窗口结束时间。
• 条件T2：在[窗口开始时间，窗口结束时间)中有数据存在。

allowedLateness

1.连接数据库

psql -h x.x.x.x -d postgres -P pager=off

create database mydb; 

\c mydb

2.创建流对象

CREATE STREAM instructor (id text, name text, dept_name text, salary float);

CREATE STREAM super_instructor AS SELECT * FROM instructor WHERE salary >= 80000 EMIT CHANGES;

demodb=# \dt
             List of relations
 Schema |       Name       |  Type  | Owner 
--------+------------------+--------+-------
 public | instructor       | stream | demo
 public | super_instructor | stream | demo
(2 rows)

3.删除流对象

DROP STREAM super_instructor;

4.往流对象插入数据

INSERT INTO instructor VALUES('10101', 'Srinivasan', 'Comp. Sci.', 65000);
INSERT INTO instructor VALUES('12121', 'Wu', 'Finance', 90000);
INSERT INTO instructor VALUES('15151', 'Mozart', 'Music', 40000);
INSERT INTO instructor VALUES('22222', 'Einstein', 'Physics', 95000);
INSERT INTO instructor VALUES('32343', 'El Said', 'History', 60000);

5.查询流对象

• PULL模式把流对象当做表对象处理，查询语法和表的查询一样。
• PUSH模式查询，流对象的查询长期运行着，一旦有增量数据立即执行后续的查询操作。

SELECT * FROM instructor;

SELECT * FROM instructor WHERE salary >= 80000 EMIT CHANGES;

6.修改流对象数据

SELECT pg_state(xmax),* FROM instructor WHERE salary >= 80000 EMIT CHANGES;

UPDATE instructor SET salary = 98000 WHERE id='22222';

7.删除流对象数据

DELETE FROM instructor WHERE id='22222';

8.常规聚合

SELECT count(id), sum(salary), dept_name 
FROM instructor 
GROUP BY dept_name 
EMIT CHANGES;

CREATE STREAM dept_cost 
AS 
SELECT count(id), sum(salary), dept_name 
FROM instructor 
GROUP BY dept_name EMIT CHANGES;

9.窗口聚合

SELECT expression [ [ AS ] output_name ] [, ...] ,
  window_begin(*) [[ AS ] output_name, 
 window_end(*) [[AS] output_name] 
FROM <stream_name> [ WHERE condition ] 
GROUP BY grouping_element 
[ HAVING condition ] 
[ { TUMBLE ( event_time_field, <window_size> ) | HOP ( event_time_field, <window_size>, <slide> ) } 
[ { WATERMARK | DELAY } <watermark_size>] 
EMIT CHANGES]
<window_size> <slide> <watermark_size> are interval 'quantity unit'

CREATE STREAM instructor2 (
id text, 
name text, 
dept_name text, 
salary float, 
evt_time timestamp with time zone default now()
);

SELECT count(id), sum(salary), avg(salary), dept_name, window_begin(*), window_end(*) 
FROM instructor2 
GROUP BY dept_name 
TUMBLE('evt_time', INTERVAL '10 seconds') 
WATERMARK INTERVAL '5s' 
EMIT CHANGES;

INSERT INTO instructor2(id, name, dept_name, salary) VALUES('33456', 'Gold', 'Physics', 87000);

SELECT pg_sleep(3);

INSERT INTO instructor2(id, name, dept_name, salary) VALUES('45565', 'Katz', 'Comp. Sci.', 75000);

SELECT pg_sleep(11);

INSERT INTO instructor2(id, name, dept_name, salary) VALUES('22222', 'Einstein', 'Physics', 95000);

CREATE STREAM instructor3 (
id text, name text, 
dept_name text, 
salary float, 
evt_time timestamp with time zone default now()
);

SELECT count(id), sum(salary), avg(salary), dept_name, window_begin(*), window_end(*) 
FROM instructor3 
GROUP BY dept_name 
HOP('evt_time', INTERVAL '2 seconds', INTERVAL '3 seconds')  
EMIT CHANGES;

INSERT INTO instructor3(id, name, dept_name, salary) VALUES('001', 'name1', 'dept1', 100);
SELECT pg_sleep(1);
INSERT INTO instructor3(id, name, dept_name, salary) VALUES('002', 'name2', 'dept1', 200);
SELECT pg_sleep(1);
INSERT INTO instructor3(id, name, dept_name, salary) VALUES('003', 'name3', 'dept1', 300);
SELECT pg_sleep(1);
INSERT INTO instructor3(id, name, dept_name, salary) VALUES('004', 'name4', 'dept1', 400);
SELECT pg_sleep(1);
INSERT INTO instructor3(id, name, dept_name, salary) VALUES('005', 'name5', 'dept1', 500);
...

10.流对象和表对象JOIN

CREATE TABLE department(id text, dept_name text, building text, budget float);

INSERT INTO department VALUES('01', 'Biology', 'Watson', 90000), ('02', 'Comp. Sci.', 'Taylor', 100000), ('03', 'Elec. Eng.', 'Taylor', 85000), ('04', 'Finance', 'Painter', 120000), ('05', 'History', 'Painter', 50000), ('06', 'Music', 'Packard', 80000), ('07', 'Physics', 'Watson', 70000);

CREATE STREAM instructor3 (id text, name text, dept_name text, salary float);

SELECT instructor3.id, name,salary, department.dept_name, building FROM instructor3 JOIN department ON instructor3.dept_name = department.dept_name EMIT CHANGES;

INSERT INTO instructor3 VALUES('12121', 'Wu', 'Finance', 90000);
INSERT INTO instructor3 VALUES('15151', 'Mozart', 'Music', 40000);