使用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>;