大数据-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 
相关文章
|
3天前
|
SQL 分布式计算 NoSQL
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
7 1
大数据-164 Apache Kylin Cube优化 案例1 定义衍生维度与对比 超详细
|
3天前
|
存储 大数据 分布式数据库
大数据-165 Apache Kylin Cube优化 案例 2 定义衍生维度及对比 & 聚合组 & RowKeys
大数据-165 Apache Kylin Cube优化 案例 2 定义衍生维度及对比 & 聚合组 & RowKeys
9 1
|
2月前
|
存储 消息中间件 Java
Apache Flink 实践问题之原生TM UI日志问题如何解决
Apache Flink 实践问题之原生TM UI日志问题如何解决
41 1
|
1月前
|
SQL 消息中间件 关系型数据库
Apache Doris Flink Connector 24.0.0 版本正式发布
该版本新增了对 Flink 1.20 的支持,并支持通过 Arrow Flight SQL 高速读取 Doris 中数据。
|
2月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
132 2
|
2月前
|
消息中间件 分布式计算 Hadoop
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
39 3
|
2月前
|
消息中间件 运维 Kafka
Apache Flink 实践问题之达到网卡的最大速度如何解决
Apache Flink 实践问题之达到网卡的最大速度如何解决
39 2
|
2月前
|
消息中间件 前端开发 Kafka
【Azure 事件中心】使用Apache Flink 连接 Event Hubs 出错 Kafka error: No resolvable bootstrap urls
【Azure 事件中心】使用Apache Flink 连接 Event Hubs 出错 Kafka error: No resolvable bootstrap urls
|
28天前
|
消息中间件 资源调度 API
Apache Flink 流批融合技术介绍
本文源自阿里云高级研发工程师周云峰在Apache Asia Community OverCode 2024的分享,内容涵盖从“流批一体”到“流批融合”的演进、技术解决方案及社区进展。流批一体已在API、算子和引擎层面实现统一,但用户仍需手动配置作业模式。流批融合旨在通过动态调整优化策略,自动适应不同场景需求。文章详细介绍了如何通过量化指标(如isProcessingBacklog和isInsertOnly)实现这一目标,并展示了针对不同场景的具体优化措施。此外,还概述了社区当前进展及未来规划,包括将优化方案推向Flink社区、动态调整算子流程结构等。
331 31
Apache Flink 流批融合技术介绍
|
3天前
|
分布式计算 监控 大数据
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
16 1

推荐镜像

更多