### --- 流式构建~~~     实时数据更新是一种普遍的需求，快速分析变化趋势才能做出正确的决策。
~~~     Kylin V1.6 发布了可扩展的 streaming cubing 功能，
~~~     它利用 Hadoop 消费 Kafka 数据的方式构建 cube。
~~~     这种方式构建的Cube能满足分钟级的更新需求。

### --- 构建流程~~~     步骤：项目 => 定义数据源(Kafka) => 定义Model
~~~     => 定义Cube => Build Cube => 作业调度(频率高)

### --- 准备数据源~~~     从Kafka消费消息，每条消息都需要包含：维度信息、度量信息、业务时间戳。
~~~     每条消息的数据结构都应该相同，
~~~     并且可以用同一个分析器将每条消息中的维度、度量和时间戳信息提取出来。
~~~     目前默认的分析器为：org.apache.kylin.source.kafka.TimedJsonStreamParser

### --- 在kafka中准备数据源~~~     # 启动kafka服务
[root@hadoop01 ~]# kafka-server-start.sh -daemon /opt/yanqi/servers/kafka_2.12/config/server.properties
~~~     # 创建名为kylin_streaming_topic的topic，有三个分区，1个副本
[root@hadoop01 ~]# kafka-topics.sh --create --zookeeper hadoop01:2181/myKafka --replication-factor 1 --partitions 3 --topic kylin_streaming_topic1

~~~     # 使用工具，每秒会向以上topic每秒发送100条记录
[root@hadoop02 ~]# kylin.sh org.apache.kylin.source.kafka.util.KafkaSampleProducer --topic kylin_streaming_topic1 --broker hadoop01:9092,hadoop02:9092~~~     # 检查消息是否成功发送
[root@hadoop02 ~]# kafka-console-consumer.sh --bootstrap-server hadoop01:9092,hadoop02:9092 --topic kylin_streaming_topic1 --from-beginning

### --- 查看通过kylin.sh加载的数据~~~     # 提取到一条，是json格式的数据，已进行格式化了
{"country": "AUSTRALIA","amount": 54.081247423872725,"qty": 9,"currency": "USD","order_time": 1637319955203,"category": "CLOTH","device": "iOS","user": {"gender": "Male","id": "5d1a9877-37e1-07e0-8103-8a2804f6fa6b","first_name": "unknown","age": 21}
}

### --- 定义流式数据源~~~     # 1、定义流式数据源：Add Streaming Table
~~~     # 2、定义Schema
~~~     # 3、定义kafka信息
~~~     # 4、查看定义的流式数据源