大数据-162 Apache Kylin 全量增量Cube的构建 Segment 超详细记录 多图

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

目前已经更新到了：

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

章节内容

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

• 构建Cube 按照日期、区域、产品、渠道
• Cube 优化方案

增量 Cube

• 在大多数业务场景下，Hive中的数据处于不断增长的状态
• 为了支持在构建Cube，无需重复处理历史数据，引入增量构建功能

Segment

Kylin将Cube划分为多个Segment（对应就是HBase中的一个表）

• 一个Cube可能由1个或多个Segment组成，Segment是指定时间范围的Cube，可以理解为Cube的分区
• Segment是针对源数据中的某个片段计算出来的Cube数据，代表一段时间内源数据的预计计算结果
• 每个Segment用起始时间和结束时间来标志
• 一个Segment的起始时间等于它之前Segment的结束前时间，它的结束时间等于它后面那个Segment的起始时间
• 同一个Cube下不同的Segment除了背后的源数据不同之外，其他如结构定义、构建过程、优化方法、存储方式等完全相同

Segment示意图

例如：以下为针对某个Cube的Segment

全量构建与增量构建

全量构建

在全量构建中：

• Cube中存在唯一一个Segment
• 每Segment没有分割时间的概念，即没有起始时间和结束时间
• 对于全量构建来说，每当需要更新Cube数据时，它不会区分历史数据和新加入的数据，即在构建时导入并处理所有的数据

增量构建

在增量构建中：

• 只会导入新Segment指定的时间区间内的原始数据，并只对这部分原始数据进行预计算

相互对比

全量构建与增量构建的Cube查询的方式对比：
全量构建Cube：

• 查询引擎只需要向存储引擎访问单个Segment所对应的数据，无需进行Segment之间的聚合
• 为了加强性能，单个Segment的数据也有可能被分片存储到引擎的多个分区上，查询引擎可能仍然需要对单个Segment不同分区的数据进一步聚合

增量构建Cube：

• 由于不同的时间的数据分布在不同的Segment中，查询引擎需要向存储引擎请求读取各个Segment的数据
• 增量构建的Cube上的查询会比全量构建的做更多的运行时聚合，通常来说增量构建的Cube上查询会比全量构建的Cube上的查询要慢一些

对于小数据量的Cube，或者经常需要全表更新的Cube，使用全量构建需要更少的运维精力，以少量的重复计算降低生产环境中的维护复杂度。
对于大数据量的Cube，例一个包含较长历史数据的Cube，如果每天更新，那么大量的资源是在用于重复计算，这个情况下可以考虑使用增量构建。

增量构建Cube过程

指定分割时间列

增量构建Cube的定义必须包含一个时间维度，用来分割不同的Segment，这样的维度称为分割时间列（Partition Date Column）。

增量构建过程

• 在进行增量构建时，将增量部分的起始时间和结束时间作为增量构建请求的一部分提交给Kylin的任务引擎
• 任务引擎会根据起始时间和结束时间从Hive中抽取相应时间的数据，并对这部分数据做预处理计算
• 将预计算的结果封装成一个新的Segment，并将相应的信息保存到元数据和存储引擎中，一般来说，增量部分的起始时间等于Cube中最后一个Segment的结束时间

增量Cube构建

步骤：定义数据源 => 定义Model => 定义Cube => 构建Cube

SQL 语句

-- 数据结构类似，只是改为了分区表
drop table wzk_kylin.dw_sales1;
create table wzk_kylin.dw_sales1(id string,channelId string,productId string,regionId string,amount int,price double
)
partitioned by (dt string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';-- 加载数据
load data local inpath "dw_sales20240101_data.txt"
into table wzk_kylin.dw_sales1
partition(dt="2024-01-01");
load data local inpath "dw_sales20240102_data.txt"
into table wzk_kylin.dw_sales1
partition(dt="2024-01-02");
load data local inpath "dw_sales20240103_data.txt"
into table wzk_kylin.dw_sales1
partition(dt="2024-01-03");
load data local inpath "dw_sales20240104_data.txt"
into table wzk_kylin.dw_sales1
partition(dt="2024-01-04");

生成数据

同样，我们先编写一个脚本来生成对应的数据：

import random# 设置参数
dates = ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04"]
num_records_per_file = 100# 定义可能的值
channel_ids = ['C001', 'C002', 'C003', 'C004']
product_ids = ['P001', 'P002', 'P003', 'P004']
region_ids = ['R001', 'R002', 'R003', 'R004']# 生成数据
for dt in dates:output_file = f'dw_sales{dt.replace("-", "")}_data.txt'with open(output_file, 'w') as f:for i in range(num_records_per_file):record_id = f"{i+1:04d}"channel_id = random.choice(channel_ids)product_id = random.choice(product_ids)region_id = random.choice(region_ids)amount = random.randint(1, 100)price = round(random.uniform(10.0, 500.0), 2)line = f"{record_id},{channel_id},{product_id},{region_id},{amount},{price}\n"f.write(line)print(f"{num_records_per_file} records have been written to {output_file}")print("All data files have been generated.")

执行的结果如下图所示：

上传数据

通过你习惯的方式，将这几个txt上传到服务器上，准备执行：

执行脚本

hive -f kylin_partition.sql

执行结果如下图：

加载数据源

Load Table From Tree

选择刚才创建的表，wzk_kylin.dw_sales1：

定义Model

增量构建的Cube需要指定分割时间列，例如：将日期分区字段添加到维度列中：
Data Model：New Join Condition，需要配置好几个：

配置成如下的结果：

维度配置如下图所示：

度量选择 AMOUNT 和 PRICE,最后的设置：

定义Cube

填写名字等跳过，维度需要添加 DT、其他都要：

配置完的结果如下图：

度量配置如下：（Bulk Add Measures 快速配置）

剩余的信息都默认填写即可：

构建Cube

接下来构建Cube的时候，进行Build：

选部分的日期，就不选所有数据了：

继续等待构建完毕：

查看Segment

刚才我们构建了

• 2024-01-01 到 2024-01-02 的数据
• 我们继续build 2024-01-02 到 2024-01-03
• 完成后继续build 2024-01-03 到 2024-01-04
  分段的进行build的任务，最后我们查看 Segment如下：

2024-01-01 到 2024-01-02 完成之后，我们继续任务：

2024-01-02 到 2024-01-03 完成之后，我们继续任务：

漫长等待，任务都完成之后如下图所示：

查询测试

第一部分：按日期和地区汇总销售数据

-- 第一部分查询：按日期和地区汇总销售数据
SELECT t1.dt,t2.regionname,SUM(t1.price) AS total_money,SUM(t1.amount) AS total_amount,MAX(t1.price) AS max_price,MIN(t1.amount) AS min_amount
FROM dw_sales1 t1
JOIN dim_region t2 
ON t1.regionid = t2.regionid
GROUP BY t1.dt, t2.regionname
ORDER BY t1.dt;

运行的结果如下图所示：

另一部分：按日期、地区和产品汇总销售数据

-- 第二部分查询：按日期、地区和产品汇总销售数据
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;

查询结果如下图所示：