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

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务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的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
目录
相关文章
|
1月前
|
Java 大数据 数据库连接
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
29 2
大数据-163 Apache Kylin 全量增量Cube的构建 手动触发合并 JDBC 操作 Scala
|
1月前
|
SQL 分布式计算 NoSQL
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
28 1
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
|
1月前
|
存储 数据挖掘 数据处理
Apache Paimon 是一款高性能的数据湖框架,支持流式和批处理,适用于实时数据分析
【10月更文挑战第8天】随着数据湖技术的发展,越来越多企业开始利用这一技术优化数据处理。Apache Paimon 是一款高性能的数据湖框架,支持流式和批处理,适用于实时数据分析。本文分享了巴别时代在构建基于 Paimon 的 Streaming Lakehouse 的探索和实践经验,包括示例代码和实际应用中的优势与挑战。
63 1
|
1月前
|
存储 大数据 分布式数据库
大数据-165 Apache Kylin Cube优化 案例 2 定义衍生维度及对比 & 聚合组 & RowKeys
大数据-165 Apache Kylin Cube优化 案例 2 定义衍生维度及对比 & 聚合组 & RowKeys
33 1
|
3月前
|
存储 消息中间件 Java
Apache Flink 实践问题之原生TM UI日志问题如何解决
Apache Flink 实践问题之原生TM UI日志问题如何解决
45 1
|
28天前
|
SQL Java API
Apache Flink 2.0-preview released
Apache Flink 社区正积极筹备 Flink 2.0 的发布,这是自 Flink 1.0 发布以来的首个重大更新。Flink 2.0 将引入多项激动人心的功能和改进,包括存算分离状态管理、物化表、批作业自适应执行等,同时也包含了一些不兼容的变更。目前提供的预览版旨在让用户提前尝试新功能并收集反馈,但不建议在生产环境中使用。
575 13
Apache Flink 2.0-preview released
|
1月前
|
存储 缓存 算法
分布式锁服务深度解析:以Apache Flink的Checkpointing机制为例
【10月更文挑战第7天】在分布式系统中,多个进程或节点可能需要同时访问和操作共享资源。为了确保数据的一致性和系统的稳定性,我们需要一种机制来协调这些进程或节点的访问,避免并发冲突和竞态条件。分布式锁服务正是为此而生的一种解决方案。它通过在网络环境中实现锁机制,确保同一时间只有一个进程或节点能够访问和操作共享资源。
63 3
|
2月前
|
SQL 消息中间件 关系型数据库
Apache Doris Flink Connector 24.0.0 版本正式发布
该版本新增了对 Flink 1.20 的支持,并支持通过 Arrow Flight SQL 高速读取 Doris 中数据。
|
3月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
217 2
|
3月前
|
消息中间件 分布式计算 Hadoop
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
51 3

推荐镜像

更多