為幫助更快掌握 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;
```

加杠杆炒股什么意思-可以杠杆的证券有哪些-【东方资本】,中保金资产管理有限公司招聘,股票杠杆交易平台有哪些,炒股怎么加杠杆操作