【大数据开发运维解决方案】Kylin消费Kafka数据流式构建cube

简介: 文章开始之前先说明环境情况,这里kylin消费的kafka数据是从Oracle 数据库用Ogg For Bigdata以json格式将数据投递到kafka topic的,投递的时候,关于insert和update 之前的数据投递到名为 ZTVOUCHER_INS 的topic,而delete和update之后的数据投递到名为 ZTVOUCHER_DEL 的topic中,这里主要介绍kylin如何消费数据创建流式cube。

文章开始之前先说明环境情况,这里kylin消费的kafka数据是从Oracle 数据库用Ogg For Bigdata以json格式将数据投递到kafka topic的,投递的时候,关于insert和update 之前的数据投递到名为 ZTVOUCHER_INS 的topic,而delete和update之后的数据投递到名为 ZTVOUCHER_DEL 的topic中,这里主要介绍kylin如何消费数据创建流式cube。

一、源端做DML操作

1.源端表ztvoucher目前没有数据,现在做insert,并查询:


insert into ztvoucher (MANDT, GJAHR, BUKRS, BELNR, BUZEI, MONAT, BUDAT, HKONT, DMBTR, ZZ0014)
values ('666', '2222', '3432', '2200001414', '001', '01', '20190101', '9101000000', 100.00, '101');

1 row created.

SQL> commit;

Commit complete.

SQL> alter system switch logfile;

System altered.
SQL>  select * from ztvoucher;
MANDT    GJAHR    BUKRS    BELNR    BUZEI    MONAT    BUDAT    HKONT    DMBTR    ZZ0014
666    2222    3432    2200001414    001    01    20190101    9101000000    100.00    101

2.去kafka查看:

[root@hadoop kafka]# ./console.sh 
input topic:ZTVOUCHER_INS
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]
.{"table":"SCOTT.ZTVOUCHER_INS","op_type":"I","op_ts":"2019-05-22 16:18:58.353767","current_ts":"2019-05-22T16:19:11.352000","pos":"00000000080000012086","tokens":{"TKN-OP-TYPE":"INSERT"},"a
fter":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":100.00,"ZZ0014":"101"}}

发现源端做的insert已经在 topic:ZTVOUCHER_INS有了。
3.源端做update操作:

update ztvoucher set dmbtr=50 where mandt='666';
commit;
alter system switch logfile;

4.去kafka查看:
先看ZTOVOUCHER_INS 内容:

[root@hadoop kafka]# ./console.sh 
input topic:ZTVOUCHER_INS
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]
.{"table":"SCOTT.ZTVOUCHER_INS","op_type":"I","op_ts":"2019-05-22 16:18:58.353767","current_ts":"2019-05-22T16:19:11.352000","pos":"00000000080000012086","tokens":{"TKN-OP-TYPE":"INSERT"},"a
fter":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":100.00,"ZZ0014":"101"}}{"table":"SCOTT.ZTVOUCHER_INS","op_type":"I","op_ts":"2019-05-22 16:22:48.354189","current_ts":"2019-05-22T16:23:33.799000","pos":"00000000080000012613","tokens":{"TKN-OP-TYPE":"SQL COMPUPD
ATE"},"after":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":50.00,"ZZ0014":"101"}}

发现除了之前的insert操作,现在update之后的数据也进来了。
再看ZTVOUCHER_DEL:

[root@hadoop kafka]# ./console.sh 
input topic:ZTVOUCHER_DEL
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]
.{"table":"SCOTT.ZTVOUCHER_DEL","op_type":"I","op_ts":"2019-05-22 16:22:48.354189","current_ts":"2019-05-22T16:23:23.781000","pos":"00000000080000012345","tokens":{"TKN-OP-TYPE":"SQL COMPUPD
ATE"},"after":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":100.00,"ZZ0014":"101"}}

发现DEL的topic中也存入了update之前的数据。
5.源端做delete操作:

delete from ztvoucher where mandt='666';
commit;
alter system switch logfile;

6.去kafka查看:

[root@hadoop kafka]# ./console.sh 
input topic:ZTVOUCHER_DEL
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]
.{"table":"SCOTT.ZTVOUCHER_DEL","op_type":"I","op_ts":"2019-05-22 16:22:48.354189","current_ts":"2019-05-22T16:23:23.781000","pos":"00000000080000012345","tokens":{"TKN-OP-TYPE":"SQL COMPUPD
ATE"},"after":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":100.00,"ZZ0014":"101"}}{"table":"SCOTT.ZTVOUCHER_DEL","op_type":"I","op_ts":"2019-05-22 16:26:26.353705","current_ts":"2019-05-22T16:27:15.049000","pos":"00000000080000012857","tokens":{"TKN-OP-TYPE":"DELETE"},"a
fter":{"MANDT":"666","GJAHR":"2222","BUKRS":"3432","BELNR":"2200001414","BUZEI":"001","MONAT":"01","BUDAT":"20190101","HKONT":"9101000000","DMBTR":50.00,"ZZ0014":"101"}}

发现除了上面update之前的数据以外,还写入了刚做的delete操作的数据。
好了,现在数据都组织好了,现在去流式创建cube。

二、流式构建cube

流式构建cube官方连接(本人用的2.4版本):
http://kylin.apache.org/cn/docs24/tutorial/cube_streaming.html
流式构建cube需要一个类型为timestamp的时间列字段用来标识消息的时间,从前面两个topic中的json数据可以看到,op_ts字段满足这个要求。
1、用j'son数据定义一张表
先来构建

相关文章
|
人工智能 OLAP 数据处理
解锁数仓内AI流水线,AnalyticDB Ray基于多模ETL+ML提效开发与运维
AnalyticDB Ray 是AnalyticDB MySQL 推出的全托管Ray服务,基于开源 Ray 的丰富生态,经过多模态处理、具身智能、搜索推荐、金融风控等场景的锤炼,对Ray内核和服务能力进行了全栈增强。
|
SQL 运维 自然语言处理
Dataphin智能化重磅升级!编码难题一扫光,开发运维更高效!
Dataphin重磅推出三大核心智能化能力:智能代码助手提升SQL开发效率;智能运维助手实现移动化任务管理;智能分析通过自然语言生成SQL,助力数据价值释放。未来将持续开放智能ETL、安全助手等能力,助力企业构建高效、稳定的数据资产体系。
888 0
|
人工智能 运维 安全
AI大模型运维开发探索第四篇:智能体分阶段演进路线
本文探讨了智能体工程的演进历程,从最初的思维链(智能体1.0)到实例化智能体(智能体2.0),再到结构化智能体(智能体3.0),最终展望了自演进智能体(智能体4.0)。文章详细分析了各阶段遇到的问题及解决策略,如工具调用可靠性、推理能力提升等,并引入了大模型中间件的概念以优化业务平台与工具间的协调。此外,文中还提到了RunnableHub开源项目,为读者提供了实际落地的参考方案。通过不断迭代,智能体逐渐具备更强的适应性和解决问题的能力,展现了未来AI发展的潜力。
|
消息中间件 关系型数据库 MySQL
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
1658 0
|
敏捷开发 运维 数据可视化
DevOps看板工具中的协作功能:如何打破开发、测试与运维之间的沟通壁垒
在DevOps实践中,看板工具通过可视化任务管理和自动化流程,提升开发与运维团队的协作效率。它支持敏捷开发、持续交付,助力团队高效应对需求变化,实现跨职能协作与流程优化。
|
人工智能 运维 自然语言处理
首个智能体模型实测:产品、开发、运维“全包了”
2025年,AI进入“动手”时代。智谱发布新一代大模型GLM-4.5,全球排名第三、国产第一,专为智能体设计,融合推理、编码与智能体能力,实现自主规划与执行任务。通过8个Demo展示其强大能力,涵盖网页设计、课件制作、小游戏开发等,展现其“带手的脑”特性,推动AI从实验室走向真实场景。
657 0
|
消息中间件 关系型数据库 MySQL
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
本教程展示如何使用Flink CDC YAML快速构建从MySQL到Kafka的流式数据集成作业,涵盖整库同步和表结构变更同步。无需编写Java/Scala代码或安装IDE,所有操作在Flink CDC CLI中完成。首先准备Flink Standalone集群和Docker环境(包括MySQL、Kafka和Zookeeper),然后通过配置YAML文件提交任务,实现数据同步。教程还介绍了路由变更、写入多个分区、输出格式设置及上游表名到下游Topic的映射等功能,并提供详细的命令和示例。最后,包含环境清理步骤以确保资源释放。
1271 2
基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
|
人工智能 DataWorks 大数据
大数据AI一体化开发再加速:DataWorks 支持GPU类型资源
大数据开发治理平台 DataWorks 的Serverless资源组支持GPU资源类型,以免运维、按需付费、弹性伸缩的Serverless架构,将大数据处理与AI开发能力无缝融合。面向大数据&AI协同开发场景,DataWorks提供了交互式开发和分析工具Notebook。开发者在创建个人开发环境时,可以选择GPU类型的资源作为Notebook运行环境,以支持进行高性能的计算工作。本教程将基于开源多模态大模型Qwen2-VL-2B-Instruct,介绍如何使用 DataWorks Notebook及LLaMA Factory训练框架完成文旅领域大模型的构建。
1200 24
|
存储 分布式计算 Hadoop
【产品升级】Dataphin V4.4重磅发布:开发运维提效、指标全生命周期管理、智能元数据生成再升级
Dataphin V4.4版本引入了多项核心升级,包括级联发布、元数据采集扩展、数据源指标上架、自定义属性管理等功能,大幅提升数据处理与资产管理效率。此外,还支持Hadoop集群管理、跨Schema数据读取、实时集成目标端支持Hudi及MaxCompute delta等技术,进一步优化用户体验。
1485 3
【产品升级】Dataphin V4.4重磅发布:开发运维提效、指标全生命周期管理、智能元数据生成再升级
|
消息中间件 存储 缓存
kafka 的数据是放在磁盘上还是内存上,为什么速度会快?
Kafka的数据存储机制通过将数据同时写入磁盘和内存,确保高吞吐量与持久性。其日志文件按主题和分区组织,使用预写日志(WAL)保证数据持久性,并借助操作系统的页缓存加速读取。Kafka采用顺序I/O、零拷贝技术和批量处理优化性能,支持分区分段以实现并行处理。示例代码展示了如何使用KafkaProducer发送消息。