使用Kafka作為數據源，通過Quark計算引擎來處理流任務，實現實時流處理的數據插入模式。

1.TimeLyre 9.1~9.2版本支持TimeLyre產品包的Quark使用流入TimeLyre的功能

2.TimeLyre 9.3版本支持TimeLyre產品包slipstream使用流入TimeLyre的功能

從Kafka數據源實時流導入Timelyre的架構如下：

1. 流處理Quark安裝及配置

（1） 安裝Quark

選擇對應產品版本的鏡像，流任務Quark的鏡像與批任務Quark為同一個，不需要使用任何的新鏡像或者patch。

（2） 分配角色及設置依賴關系

在分配角色步驟，選擇共享已有Quark(通常為執行批任務的Quark)的元數據，如果當前Quark為唯一quark，則不用選擇共享元數據。

（3） 設置計算引擎指定模式

在配置服務步驟將ngmr.engine.mode修改為morphling

（4） 后續安裝

后續正常執行剩余的安裝步驟即可

（5） 驗證流處理Quark是否可用

檢查是否可以正常建立TimeLyre表
在配置頁面檢查ngmr.engine.mode是否為morphling
Manager的管理→許可證頁面檢查是否有Slipstream的license(最低版本Standard及以上即可)

2. TimeLyre產品的Slipstream安裝

（1） 安裝Slipstream

TimeLyre 9.3 版本新增slipstream組件使用流入TimeLyre功能

（2） 分配角色及設置依賴關系

（3） 后續安裝

后續正常執行剩余的安裝步驟即可

3. 流入TimeLyre步驟

（1）創建TimeLyre表

舉例

create table output_timelyre(
  tagname string,
  value double,
  state int,
  valtime timestamp
) stored as timelyre
TBLPROPERTIES (
'timelyre.timestamp.col'='valtime',
'timelyre.tag.cols'='tagname',
'epoch.engine.enabled'='false'
);

（2）創建Kafka數據源

在創建Kafka數據源之前，需要確認集群是否已安裝Kafka服務且服務是否處于健康狀態。

找到TDH-Client所在目錄，在TDH-Client/kafka/bin目錄下可以找到Kafka服務的應用腳本，包括創建topic、生產者與消費者使用等。同時，開發者也可以使用Kafka API進行數據的寫入與讀取。使用者可自行參照《Slipstream使用手冊》學習使用Kafka，在此不再贅述。

（3）創建Stream及觸發StreamJob

CREATE STREAM用于建一個input stream

語法：創建以Kafka為源的input stream

CREATE STREAM <stream_name>(<column> <data_type>, <column> <data_type>, ...)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '<delimiter>' (1)
TBLPROPERTIES("topic"="<topic_name>",  (2)
              "kafka.zookeeper"="<ip:port>",  (3)
              "kafka.broker.list"="<ip:port>" (4)
              ["<key>"="<value>",] (5)
              ["transwarp.consumer.<property_key>"="kafka_consumer_property_value",] (6)
              ["transwarp.producer.<property_key>"="kafka_producer_property_value",] (7)
              ...);

1	<delimiter>指定數據中的分隔符。
2	<topic_name>指定input stream指向的Kafka topic。
3	<ip:port>指定TDH集群中的一個Zookeeper節點的ip和端口號，除非另外設置，Zookeeper的端口號為2181。
4	<ip:port>指定集群中的任意一個Kafka broker的ip和端口號，默認為9092。
5	TBLPROPERTIES中還可以包含任意自定義的鍵值對用于存放用戶自定義的流屬性。
6	指定StreamJob啟動Kafka consumer使用的屬性值。在安全模式下以Kafka為源時需要配置安全相關參數，請參考《Slipstream使用手冊》。
7	當stream作為輸出流時，可通過此配置項來設定StreamJob啟動Kafka producer時使用的屬性值。

舉例

CREATE STREAM demo_stream
(
  tagname string,
  value double,
  state int,
  valtime bigint
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
TBLPROPERTIES
(
"topic"="demo_test",
"kafka.zookeeper"="timelyre-28:2181", --根據集群實際值填寫
"kafka.broker.list"="timelyre-28:9092", --根據集群實際值填寫
"transwarp.consumer.sasl.enabled.mechanisms"="GSSAPI",
"transwarp.consumer.sasl.mechanism.inter.broker.protocol"="GSSAPI",
"transwarp.consumer.security.protocol"="SASL_PLAINTEXT",
"transwarp.consumer.sasl.kerberos.service.name"="kafka",
"transwarp.consumer.sasl.jaas.config"=
"com.sun.security.auth.module.Krb5LoginModule required \
  useKeyTab=true \
  storeKey=true \
  keyTab=\"/etc/quark1/conf/kafka.keytab\" \ --根據集群實際情況填寫
  principal=\"kafka@TDH\"" --根據集群實際情況填寫
);

（4） 查看STREAM

SHOW STREAMS;
+--------------+
|   tab_name   |
+--------------+
| demo_stream  |
+--------------+

可以查詢到剛剛建好的demo_stream。

（5） 建streamjob并執行

--創建流任務
create streamjob test_job as("insert into output_timelyre select  * from demo_stream")
jobproperties (
    'morphling.job.enable.checkpoint'='true', -- 流任務啟用高可用
    'morphling.job.checkpoint.interval'='60000', -- Checkpoint的時間間隔為60秒
    'morphling.result.auto.flush'='false' --是否來一條數據就刷到輸出表中,生產上默認設置為false,測試可以開啟
);

--啟動流任務
start streamjob test_job;

可以通過如下命令查看正在運行的StreamJob并獲取streamjob_id。：

LIST STREAMJOBS;

（6） 停止Streamjob

一個Streamjob創建后，需要使用者手動停止，否則將會一直運行下去。停止Streamjob：

STOP STREAMJOB <streamjob_id>;

股票配资公司网站-炒股开杠杆-【东方资本】,杠杆型公司,股票开通配资有风险吗怎么办呢,什么叫加杠杆