大数据-166 Apache Kylin Cube 流式构建 整体流程详细记录

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 大数据-166 Apache Kylin Cube 流式构建 整体流程详细记录

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka(已更完)

Spark(已更完)

Flink(已更完)

ClickHouse(已更完)

Kudu(已更完)

Druid(已更完)

Kylin(正在更新…)

章节内容

上节我们完成了如下的内容:


Cube 的优化

案例 2 定义衍生维度及对比

聚合组 详细讲解

RowKeys 详细讲解

dd31a51e4e24cc8234a53ecd9d90e6fd_6c39366b724e4430a7291d42d9a122f1.png 基本概念

实时数据更新是一种普遍的需求,快速分析趋势才能做正确的决策。

KylinV1.6发布了扩展StreamingCubing功能,它利用Hadoop消费Kafka数据的方式构建Cube,这种方式构建的Cube能满足分钟级的更新需求。


实现步骤

步骤:项目 => 定义数据源(Kafka)=> 定义Model => 定义Cube => Build Cube => 作业调度(频率高)


生成数据

从Kafka消费消息,每条消息都需要包含:


维度信息

度量信息

业务时间戳

每条消息的数据结构都应该相同,并且可以用同一个分析器将每条消息中的维度、度量和时间戳信息提取出来。

目前默认的分析器为:org.apache.kylin.source.kafka.TimedJsonStreamParser

创建数据

# 创建名为kylin_streaming_topic的topic
kafka-topics.sh --create --zookeeper h121.wzk.icu:2181 --replication-factor 1 --partitions 1

执行结果如下图所示:

数据采样

设置采样器

kylin.sh org.apache.kylin.source.kafka.util.KafkaSampleProducer --topic kylin_streaming_topic1

发了一大批数据,如下图所示:

检查数据

检查数据是否发送成功:

kafka-console-consumer.sh --bootstrap-server h121.wzk.icu:9092 --topic kylin_streaming_topic

数据样例如下所示:

{"country":"INDIA","amount":44.47793969871658,"qty":3,"currency":"USD","order_time":1723358207350,"category":"TOY","device":"iOS","user":{"gender":"Female","id":"1c54f68e-f89a-b5d2-f802-45b60ffccf60","first_name":"unknown","age":15}}
{"country":"AUSTRALIA","amount":64.86505054935878,"qty":9,"currency":"USD","order_time":1723358207361,"category":"TOY","device":"iOS","user":{"gender":"Female","id":"de11d872-e843-19c9-6b35-9263f1d1a2a1","first_name":"unknown","age":19}}
{"country":"CANADA","amount":90.1591854077722,"qty":4,"currency":"USD","order_time":1723358207371,"category":"Other","device":"Andriod","user":{"gender":"Male","id":"4387ee8b-c8c1-1df4-f2ed-c4541cb97621","first_name":"unknown","age":26}}
{"country":"INDIA","amount":59.17956535472526,"qty":2,"currency":"USD","order_time":172335820738

定义数据源

数据源选择Add Streaming Table:

点击之后,把刚才的JSON填写进去,就可以解析出来:

定义Kafka信息,填写对应的内容,如下图所示:

可以看到我们刚才添加的内容如下图所示:

定义Model

新建Model,如下图所示,名称随意:

原则DataModel,如下图所示:

选择维度Dimension信息:

选择度量Measures,如下图所示:

设置Setting中,设置对应的PartitionDateColumn信息,如下图:

定义Cube

名字随意,自己能分清就可以,如下图:

设置Dimensions信息如下图所示:

设置度量Measure信息如下图所示:

RefreshSetting设置信息如下图所示:

设置Aggregation Groups信息:

RowKeys的设置如下图所示:

StreamingCube 和 普通的Cube大致上一样,以下几点需要注意:


分区时间列应该Cube的一个Dimension,在SteamingOLAP中时间总是一个查询条件,Kylin利用它来缩小扫描分区的范围

不要使用order time作为dimmension 因为它非常精细,建议使用minute_start、hour_start或其他,取决于用户如何查询数据

定义 year_start、quarter_start、month_start、day_start、hour_start、minute_start或其他,取决于用户如何查询数据

在RefreshSetting设置中,创建更多合并的范围,如0.5时、4小时、1天、7天,这样设置有助于控制CubeSegment的数量

在RowKeys部分,拖拽minute_start到最上面的位置,对于Streaming查询,时间条件会一直显示,将其放到前面将会缩小扫描范围。

构建Cube

可以通过 HTTP 的方式完成构建

curl -X PUT --user ADMIN:KYLIN -H "Content-Type:
application/json;charset=utf-8" -d '{ "sourceOffsetStart": 0, "sourceOffsetEnd": 92233720368

也可以使用WebUI,我比较喜欢用页面来构建:

执行查询

select minute_start, count(*), sum(amount), sum(qty) from streamingds1
group by minute_start
order by minute_start

自动构建

用 crontab 来定时任务,让其定时执行:

crontab -e */20 * * * * curl -X PUT --user ADMIN:KYLIN -H "Content-Type:application/json;cha
相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
2月前
|
人工智能 数据处理 API
阿里云、Ververica、Confluent 与 LinkedIn 携手推进流式创新,共筑基于 Apache Flink Agents 的智能体 AI 未来
Apache Flink Agents 是由阿里云、Ververica、Confluent 与 LinkedIn 联合推出的开源子项目,旨在基于 Flink 构建可扩展、事件驱动的生产级 AI 智能体框架,实现数据与智能的实时融合。
382 6
阿里云、Ververica、Confluent 与 LinkedIn 携手推进流式创新,共筑基于 Apache Flink Agents 的智能体 AI 未来
|
消息中间件 数据挖掘 Kafka
Apache Kafka流处理实战:构建实时数据分析应用
【10月更文挑战第24天】在当今这个数据爆炸的时代,能够快速准确地处理实时数据变得尤为重要。无论是金融交易监控、网络行为分析还是物联网设备的数据收集,实时数据处理技术都是不可或缺的一部分。Apache Kafka作为一款高性能的消息队列系统,不仅支持传统的消息传递模式,还提供了强大的流处理能力,能够帮助开发者构建高效、可扩展的实时数据分析应用。
726 5
|
8月前
|
存储 人工智能 数据处理
Apache Doris 2025 Roadmap:构建 GenAI 时代实时高效统一的数据底座
秉承“以场景驱动创新” 的核心理念,持续深耕三大核心场景的关键能力,并对大模型 GenAI 场景的融合应用进行重点投入,为智能时代构建实时、高效、统一的数据底座。
435 10
Apache Doris 2025 Roadmap:构建 GenAI 时代实时高效统一的数据底座
|
9月前
|
存储 分布式数据库 Apache
小米基于 Apache Paimon 的流式湖仓实践
小米基于 Apache Paimon 的流式湖仓实践
243 0
小米基于 Apache Paimon 的流式湖仓实践
|
10月前
|
存储 分布式数据库 Apache
小米基于 Apache Paimon 的流式湖仓实践
本文整理自Flink Forward Asia 2024流式湖仓专场分享,由计算平台软件研发工程师钟宇江主讲。内容涵盖三部分:1)背景介绍,分析当前实时湖仓架构(如Flink + Talos + Iceberg)的痛点,包括高成本、复杂性和存储冗余;2)基于Paimon构建近实时数据湖仓,介绍其LSM存储结构及应用场景,如Partial-Update和Streaming Upsert,显著降低计算和存储成本,简化架构;3)未来展望,探讨Paimon在流计算中的进一步应用及自动化维护服务的建设。
563 0
小米基于 Apache Paimon 的流式湖仓实践
|
存储 数据挖掘 数据处理
巴别时代使用 Apache Paimon 构建 Streaming Lakehouse 的实践
随着数据湖技术的发展,企业纷纷探索其优化潜力。本文分享了巴别时代使用 Apache Paimon 构建 Streaming Lakehouse 的实践。Paimon 支持流式和批处理,提供高性能、统一的数据访问和流批一体的优势。通过示例代码和实践经验,展示了如何高效处理实时数据,解决了数据一致性和故障恢复等挑战。
313 61
|
消息中间件 Java Kafka
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
472 1
|
存储 Cloud Native 数据处理
从嵌入式状态管理到云原生架构:Apache Flink 的演进与下一代增量计算范式
本文整理自阿里云资深技术专家、Apache Flink PMC 成员梅源在 Flink Forward Asia 新加坡 2025上的分享,深入解析 Flink 状态管理系统的发展历程,从核心设计到 Flink 2.0 存算分离架构,并展望未来基于流批一体的通用增量计算方向。
312 0
从嵌入式状态管理到云原生架构:Apache Flink 的演进与下一代增量计算范式
|
4月前
|
SQL 人工智能 数据挖掘
Apache Flink:从实时数据分析到实时AI
Apache Flink 是实时数据处理领域的核心技术,历经十年发展,已从学术项目成长为实时计算的事实标准。它在现代数据架构中发挥着关键作用,支持实时数据分析、湖仓集成及实时 AI 应用。随着 Flink 2.0 的发布,其在流式湖仓、AI 驱动决策等方面展现出强大潜力,正推动企业迈向智能化、实时化的新阶段。
544 9
Apache Flink:从实时数据分析到实时AI
|
4月前
|
SQL 人工智能 API
Apache Flink 2.1.0: 面向实时 Data + AI 全面升级,开启智能流处理新纪元
Apache Flink 2.1.0 正式发布,标志着实时数据处理引擎向统一 Data + AI 平台迈进。新版本强化了实时 AI 能力,支持通过 Flink SQL 和 Table API 创建及调用 AI 模型,新增 Model DDL、ML_PREDICT 表值函数等功能,实现端到端的实时 AI 工作流。同时增强了 Flink SQL 的流处理能力,引入 Process Table Functions(PTFs)、Variant 数据类型,优化流式 Join 及状态管理,显著提升作业稳定性与资源利用率。
491 0

推荐镜像

更多