当前位置：首页 > news >正文

大数据-163 Apache Kylin 全量增量Cube的构建手动触发合并 JDBC 操作 Scala

news 文章来源：https://blog.csdn.net/w776341482/article/details/142844747 2025/5/10 20:05:04

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

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（已更完）
ClickHouse（已更完）
Kudu（已更完）
Druid（已更完）
Kylin（正在更新…）

章节内容

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

全量增量Cube的构建
Segment

在这里插入图片描述

基本流程

在 Apache Kylin 中，手动触发 Segment 合并的步骤如下：

选择要合并的 Cube 和 Segments：进入 Kylin Web UI，选择你要操作的 Cube，进入该 Cube 的详情页面。在“Segments”标签页下，可以看到当前 Cube 的所有 Segments。选择你希望合并的 Segments。
合并 Segments：点击页面上的 “Merge Segment” 按钮。通常情况下，Kylin 会自动计算可以合并的 Segments。如果你想手动控制合并的 Segments，可以在弹出的对话框中手动选择你想合并的 Segments。
配置合并任务：配置合并任务的参数，如目标时间范围等。Kylin 会根据你选择的 Segments 的范围自动填充一些默认的值。你可以根据需求调整这些参数。
启动合并任务：完成配置后，点击 “Submit” 按钮。Kylin 将会创建一个新的合并任务（Job），该任务将在后台执行。你可以在 “Job” 页面查看任务的执行状态。
监控任务状态：在 “Job” 页面，你可以查看合并任务的日志和状态。如果任务执行成功，你会看到新的 Segment 出现在 Segments 列表中，表示合并已经完成。
如果合并成功，新的合并后 Segment 会替代原来的多个 Segments，而旧的 Segments 将被 Kylin 自动清理。

需要注意的是，手动合并的操作可能会占用大量资源，因此在高负载时需要谨慎操作，并在合适的时间段执行合并任务。

手动触发合并Segment

Kylin提供了一种简单的机制用于控制Cube中Segment的数量：合并Segment，在WebGUI中选中需要进行Segments合并的Cube。

单击Action => Merge

我们刚才分阶段进行了任务的Build操作，
01-01、01-02、01-03、01-04 的任务，我们可以使用 Merge 来进行合并：

在这里插入图片描述

选中需要合并的Segment，可以同时合并多个Segment，但这些Segment必须是连续的，单击提交系统会提交一个类型为 MERGE 的构建任务，这里可以选择时间阶段，我选择的是 01-03到01-04：
在这里插入图片描述

提交任务，可以看到是一个 Merge任务，看名字：【MERGE】，等待合并完毕：
在这里插入图片描述
合并完毕的结果如下图：

注意事项

在MERGE构建结束之前，所有选中用来合并的Segment仍然处于可用的状态
在MERGE类型的构建完成之前，系统将不允许提交这个Cube上任何类型的其他构建任务
当MERGE构建结束的时候，系统将选中合并的Segment替换为新的Segment，而被替换下的Segment等待将被垃圾回收和清理，以节省系统资源

删除Segment

使用WebUI删除Cube的Segment，
这里选择 Disable 就可以删除Segment了：

在这里插入图片描述
Disable之后，可以看到下面的：DeleteSegment操作，就可以删除指定的Segment了：

自动合并

手动维护Segment很繁琐，人工成本高，Kylin中是可以支持自动合并Segment。
在Cube Designer的 Refresh Settings的页面中有：

Auto Merge Thresholds
Retention Thresholds

Refresh Settings的页面：
在这里插入图片描述
两个设置项可以用来帮助管理Segment碎片，这两项设置搭配使用这两项设置可以大大减少对Segment进行管理的麻烦。

Auto Merge Thresholds

允许用户设置几个层级的时间阈值，层级越靠后，时间阈值越大
每当Cube中有新的Segment状态变为READY的时候，就会自动触发一次系统自动合并

合并策略

尝试最大一级的时间阈值，例如：针对（7天、28天）层级的日志，先检查能够将连续的若干个Segment合并成为一个超过28天的大Segment
如果有个别的Segment的事件长度本身已经超过28天，系统会跳过Segment
如果满足条件的连续Segment还不能够累积超过28天，那么系统会使用下一个层级的时间戳重复寻找

案例1 理解Kylin自动合并策略

假设自动合并阈值设置为7天、28天
如果现在有A-H 8个连续的Segment，它们的时间长度为28天（A）、7天（B）、1天（C）、一天（D）、一天（E）、一天（F）、一天（G）、一天（H）
此时，第9个Segment加入，时间长度为1天

自动合并的策略为：

Kylin判断时能够将连续的Segment合并到28天这个阈值，由于Segment A已经超过28天，会被排除。
剩下的连续Segment，所有时间加一起 B+C+D+E+F+G+H+I < 28天，无法满足28天的阈值，则开始尝试7天的阈值
跳过 A（28）、B（7）均超过7天，排除
剩下的连续Segment，所有时间加在一起 C+D+E+F+G+H+I 达到7天的阈值，触发合并，提交Merge任务，并构建一个SegmentX（7天）
合并后，Segment为：A（28天）、B（7天）、X（7天）
连续触发检查，A（28天）跳过，B+X（7+7=14）< 28天，不满足第一阈值，重新使用第二阈值触发
跳过B、X尝试终止

案例2 配置自动合并4天的Segment

选中Model，选择Edit进行编辑：
在这里插入图片描述
直接到Refresh Setting选项卡，将选项修改为，4天：

后续将自动进行Segment的构建。

配置保留的Segment

自动合并是将多个Segment合并为一个Segment，以达到清理碎片的目的，保留Segment则是及时清理不再使用的Segment。
在很多场景中，只会对过去一段时间内的数据进行查询，例如：

对于某个只显示过去1年数据的报表
支持它的Cube其实只需要保留过去一年类的Segment即可
由于数据在Hive中已经存在备份，则不需在Kylin中备份超过一年的类似数据

可以将Retention Threshold设置为365，每当有新的Segment状态变为READY的时候，系统会检查每一个Segment。如果它的结束时间距离最晚的一个Segment的结束时间已经大于等于RetentionThreshold，那么这个Segment将视为无需保留，系统会自动从Cube中删除这个Segment。

保留策略示意图如下所示：
在这里插入图片描述

使用JDBC连接操作Kylin

简单介绍

要将数据以可视化方式展示出来，需要使用Kylin的JDBC方式连接执行SQL，获取Kylin的执行结果
使用Kylin的JDBC与JDBC操作MySQL一致

业务需求

通过JDBC的方式，查询按照日期、区域、产品维度统计订单总额/总数量结果

开发步骤

添加依赖

<dependency><groupId>org.apache.kylin</groupId><artifactId>kylin-jdbc</artifactId><version>3.1.1</version>
</dependency>

实现规划

创建Connection连接对象
构建SQL语句
创建Statement对象，并执行executeQuery
打印结果

编写代码

我这里用Scala实现了，Java也差不多

package icu.wzk.kylinimport java.sql.DriverManagerobject KylinJdbcTest {def main(args: Array[String]): Unit = {// 创建连接对象val connection = DriverManager.getConnection("jdbc:kylin://h122.wzk.icu:7070/wzk_test_kylin", "ADMIN", "KYLIN")// 创建Statementval statement = connection.createStatement();// 构建SQL语句var sql ="""|select| t1.dt,| t2.regionid,| t2.regionname,| t3.productid,| t3.productname,| sum(t1.price) as total_money,| sum(t1.amount) as total_amount|from| dw_sales1 t1|inner join dim_region t2|on t1.regionid = t2.regionid|inner join dim_product t3|on t1.productid = t3.productid|group by| t1.dt,| t2.regionid,| t2.regionname,| t3.productid,| t3.productname|order by| t1.dt,| t2.regionname,| t3.productname|""".stripMarginval resultSet = statement.executeQuery(sql)println("dt region product_name total_money total_amount")while (resultSet.next()) {// 获取时间val dt = resultSet.getString("dt")// 获取区域名称val regionName = resultSet.getString("regionname")// 获取产品名称val productName = resultSet.getString("productname")// 获取累计金额val totalMoney = resultSet.getDouble("total_money")// 获取累计数量val totalAmount = resultSet.getDouble("total_amount")println(f"$dt $regionName $productName $totalMoney $totalAmount")}connection.close()}}

测试运行

我们运行代码，可以看到如下的运行结果：
在这里插入图片描述