为帮助更快掌握 Slipstream 的工作原理和用法，我们将通过一个简易的案例进行演示。

本案例中，我们首先将 Kafka 作为上游生成样例数据，随后在 Slipstream 中创建流任务并接收和处理来自 Kafka 的数据。

准备 Kafka 环境

登陆集群中安装了 Kafka 的节点，确保已启动 Kafka 相关组件，如 ZooKeeper 等。

Kafka 安装与环境启动方法，参见官方文档。

进入 Kafka 的安装目录（本案例为 /usr/lib/kafka/bin），然后执行下述命令创建一个名为 demo 的 Topic（拥有三个分区）。

下述环境中，我们假设 ZooKeeper 的访问地址为 172.16.1.128，您需要根据实际情况替换。

./kafka-topics.sh  --zookeeper 172.16.1.128:2181 --create --topic demo --partitions 3 --replication-factor 1

如果 Kafka 处在安全模式下，您需要提供用户的 Principal 和 Keytab，并且需确保该用户具备创建 Topic 的权限。

重新打开一个命令行窗口，登录集群中的任意节点，然后通过 Beeline 连接至 Slipstream，更多介绍，见《交互方式》章节。

以 LDAP 认证方式为例，命令示例如下。
```
beeline -u "jdbc:hive2://<server_ip/hostname>:<port>/<database_name>" -n <username> -p <password>
```
- <server_ip/hostname>：Slipstream 服务的 IP 地址或主机名。
- <port>：服务端口，默认为 10010。
- <database_name>：要连接到的数据库。
- <username>：用于身份认证的用户名。
- <password>：用户名对应的密码。
  
  连接示例如下：
beeline -u "jdbc:hive2://172.16.20.1:10010/demodata" -n core_user_01 -p Slipstream@123

依次执行下述命令，创建输入流、输出表和流任务。

-- 创建输入流，添加我们准备的 Kafka 数据源，指定接收到的消息按逗号分隔为 id、name、amt 和 ts 四列。
CREATE STREAM input_stream (
id string,name string,amt bigint,ts timestamp
)ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
TBLPROPERTIES("topic"="demo","kafka.zookeeper"="172.16.1.128:2181","kafka.broker.list"="172.16.1.128:9092");

-- 创建输出表，即从 Kafka 接收到的数据会写入至该表中
create table etl_output(id string, name string,amt bigint,ts timestamp);

-- 创建流任务
create streamjob simple_etl as ("insert into etl_output select * from input_stream")
jobproperties("morphling.result.auto.flush"="true");
-- 注意：这里 morphling.result.auto.flush=true 表示每接收到一条数据就刷到输出表，在测试场景下可快速获取输入源数据，如您用于生产环境不建议开启该功能。

执行下述命令启动流任务。

start streamjob simple_etl;

任务启动后，流任务将从 Kafka 中拉取当前时刻开始的消息数据，默认情况下，不会拉取任务启动之前的消息数据。

切换至 Kafka 所属机器的命令窗口，执行下述命令启动 Kafka Producer，并输入几条样例消息数据。

kafka-console-producer.sh --broker-list 172.16.1.128:9092 --topic demo

1,name1,10,2021-10-10 10:11:01
2,name2,20,2021-10-10 10:11:02
3,name3,30,2021-10-10 10:11:03

回到 Slipstream 的 Beeline 命令行，执行下述命令查看输出表的内容。

select * from etl_output order by id;

返回结果如下，可查看到数据已完成实时同步。

+-----+--------+------+------------------------+
| id  |  name  | amt  |           ts           |
+-----+--------+------+------------------------+
| 1   | name1  | 10   | 2021-10-10 10:11:01.0  |
| 2   | name2  | 20   | 2021-10-10 10:11:02.0  |
| 3   | name3  | 30   | 2021-10-10 10:11:03.0  |
+-----+--------+------+------------------------+

（可选）如不再需要运行流任务，可执行下述命令停止流任务。
```
stop streamjob simple_etl;
```