---

Confluent:5.3.1

Confluent安装：https://blog.csdn.net/u012551524/article/details/103686739

kafka:2.0.0

---

1、修改配置文件/data/software/confluent-5.3.1/etc/ksql/ksql-server.properties

bootstrap.servers=bigdata-dev1:9092,bigdata-dev2:9092,bigdata-dev3:9092

2、启动ksql server

/data/software/confluent-5.3.1/bin/ksql-server-start   /data/software/confluent-5.3.1/etc/ksql/ksql-server.properties

3、启动KSQL CLI

LOG_DIR=/tmp/ksql_logs /data/software/confluent-5.3.1/bin/ksql http://localhost:8088

4、table和stream的区别

stream:结构化数据的序列，一旦事件被引入到流是不可变的，这意味着它不能被更新或删除。

table:基于从流过来的事件目前的局势。例如，ProductA的总库存量是多少？表中的事实是可变的，如果ProductA不再有库存，则可以更新或删除ProductA的数量。

5、创建一个stream

CREATE STREAM action_log
(ID STRING, 
WRITE_DATE STRING, 
NAME STRING) with 
(kafka_topic='action.log',value_format='DELIMITED');

6、查看stream action_log

select id,WRITE_DATE,NAME from action_log;

7、创建一个producer向topic action.log写测试数据

8、ksql cli端查看

9、创建一张表，统计基于用户的count

create table action_log_count as \
select NAME, count(ID) action_count \
from action_log \
group by NAME

10、查看表内容