开发动态表实现近实时增量处理

教程概述

通过本教程，你将了解到如何使用Lakehouse动态表(Dynamic Table)构建一个完整的实时数据ETL加工流程。场景设计如下：

本教程将通过数据导入任务、基于动态表实现的数据清洗转换和数据聚合任务，实现了一个分钟级延迟的流式数据处理Pipeline案例。

本教程借助 Lakehouse Tutorial，提供了在线教程指南以及脚本导入功能。你可以在登录 Lakehouse Web 控制台后，通过 Lakehouse Tutorial 的“Lakehouse教程”入口进入，并根据在线指南引导完成教程。

入门知识

动态表（Dynamic Table）是一种支持仅处理增量变化数据的对象类型，与物化视图类似。创建动态表时，你需要同时定义数据计算逻辑。通过刷新操作，可以触发动态表的计算逻辑并更新其数据。与传统的 ETL 任务相比，动态表只需要在定义时声明全量语义的计算逻辑，在刷新时则可自动进行增量计算优化。借助动态表，Lakehouse可以面向流式写入的数据进行近实时增量处理，为实时分析提供数据准备。

教程步骤

数据导入：创建原始表、持续写入用户行为数据；
开发动态表模型：通过 Dynamic Table 分别构造数据清洗、数据聚合的加工流程。
验证加工结果：监控由动态表构造的数据Pipeline的刷新执行状态，同时查看消费层数据表的变化情况，验证实时数据加工结果

通过以上步骤，你将了解如何通过动态表在 Lakehouse 中开发流式处理任务。

准备工作

首先，创建行为日志表，通过INSERT INTO任务写入测试数据。

为了简化教程，本教程通过周期性调度 INSERT INTO 任务模拟数据的持续写入。实际生产环境中往往使用流式写入接口实时接入数据，云器 Lakehouse 也提供 SDK/Flink Connector 等方式进行实时写入，不建议在生产任务中使用 INSERT INTO 方式写入。

本教程在“开发”模块提供了创建数据表以及插入测试数据的 SQL 脚本。打开 [Tutorial_Working_With_Dynamic_Table->Step01.Preparation] 脚本文件，配置 1Min 间隔的调度策略并提交部署，以模拟实时数据导入。

Transform Layer动态表模型开发

在“开发”模块打开 [Tutorial_Working_With_Dynamic_Table->Step02.Transformation_With_Dynamic_Table] 样例动态表任务文件。该动态表定义了对原始表进行清洗转换的 ETL 逻辑。

请参考下图进行调度配置"运行集群"和"调度参数"：

提交部署

点击[提交]按钮，将动态表模型部署到Lakehouse目标环境中，系统将根据动态表设置的刷新频率自动刷新。

Aggregate Layer动态表模型开发

在“开发”模块打开 [Tutorial_Working_With_Dynamic_Table->Step03.Aggregation_With_Dynamic_Table] 样例动态表任务文件。该动态表定义了对中间层数据模型进行聚合分析的 ETL 逻辑。

请参考下图进行调度配置和提交运行：

验证增量更新结果

在“开发”模块打开 Step04.Check_Data_Freshness 文件，通过执行 Query 检查自动刷新动态表的数据新鲜度。

如上图所示，借助动态表的自动增量刷新功能，能够以 1min 左右的时间延迟完成数据的实时处理。

环境清理

在”任务运维”模块的周期性任务列表中，下线数据导入任务；在动态表任务列表中，下线 2 个动态表模型任务。

常见问题

Q1：动态表刷新后数据没有更新，怎么排查？

执行
```
SHOW DYNAMIC TABLE REFRESH HISTORY
```
SHOW DYNAMIC TABLE REFRESH HISTORY
查看最近刷新状态，确认刷新是否成功执行
确认源表是否有新数据写入（用
```
SELECT COUNT(*)
```
SELECT COUNT(*)
对比刷新前后的行数）
检查刷新间隔设置是否合理，
```
INTERVAL
```
INTERVAL
太长会导致数据延迟
如果
```
refresh_mode
```
refresh_mode
显示
```
FULL
```
FULL
而非
```
INCREMENTAL
```
INCREMENTAL
，说明 SQL 中包含不支持增量计算的算子

Q2：刷新报错 “incremental refresh not supported”，怎么办？

某些 SQL 算子不支持增量计算，例如带
```
ORDER BY
```
ORDER BY
的窗口函数、某些子查询
解决方案：简化 SQL 逻辑，或接受全量刷新，使用
```
CREATE OR REPLACE DYNAMIC TABLE ... REFRESH_MODE = FULL
```
CREATE OR REPLACE DYNAMIC TABLE ... REFRESH_MODE = FULL
强制全量模式

Q3：选哪个 VCluster 来执行刷新？

刷新是计算密集型操作，建议使用 GENERAL 类型集群
不要和查询业务共用同一个集群，避免刷新任务影响查询响应时间
刷新频率高（< 1 分钟）时，建议使用专用的小规格集群（XSMALL）

Q4：动态表和普通 Studio 定时任务相比，什么时候用哪个？

场景	推荐方案
纯 SQL 增量加工，依赖关系简单	Dynamic Table（自动管理依赖和刷新）
需要 Python 处理、调用外部 API	Studio 任务
多个上下游任务有复杂依赖编排	Studio 任务流
需要精确控制执行时间（如整点）	Studio 定时任务

联系我们