摘要：本文整理自阿里云智能集团 、Apache Flink Committer 刘大龙老师在 2024FFA 流批一体论坛的分享。内容主要为以下三个部分：

一、A User Story Of Data Engineer

二、Materialized Table 构建流批一体 ETL

三、Demo

一、A User Story Of Data Engineer

如今，数据湖（Data Lake）和湖仓一体（Lakehouse）的概念，以及像 Paimon 这样的技术栈，正变得越来越流行。理想的解决方案是以开放的数据湖格式作为核心，这种做法允许将多样化的数据源导入到统一的数据存储中。由于采用了开放格式，因此能够支持多种计算引擎，包括流处理、批处理、OLAP 分析甚至是 AI 模型训练等任务。通过这种方式，可以对数据执行多维度的分析与处理，并最终将结果应用于人工智能应用、商业智能报告或其他数据分析场景中，从而帮助企业更好地挖掘其数据资产的价值。

虽然这一架构看起来非常吸引人，并且很多组织已经在尝试实施，但在实际操作过程中可能会遇到一系列挑战。

接下来我们可以从一个案例看一下：假设 Steven 是某公司的数据工程师，上班之后接收到老板的钉钉消息——“能不能统计下平台昨天的一个热销商品，或者一些各品类的 GMV？”，熟悉业务的 Steven 选择 Iceberg 作为它的数据湖存储，然后把数据从 MySQL 里面通过 DataX 导进来，导完之后用 Spark 做 Batch ETL ，然后通过 QuickBI 搭建一个很漂亮的大屏，老板看到大屏之后很满意。然后提出一个新的需求。

就是实现报表每天更新，接到需求后 Steven 在现有的批链路上加一个调度器，比如用 Airflow，实现每天调度批作业，实现了报表的每天更新。

于是老板又期望报表每天都实时更新，Steven 对 Iceberg 比较熟，知道 Iceberg 当前主要面向离线场景，在实时更新上可能有一些困难。Spark 或者 Spark SQL 很难做到一个秒级或分钟级的实时计算，在现有的统一的湖仓架构的链路上无法满足，这时候不得不搭建一套新的实时链路。比如用 Flink 加 Kafka 典型的组合完成一套实时的链路，然后重新搭建一套实时的 BI 大屏。这时候相当于引入实时和离线的两套链路，本质上就是一个 Lambda 的架构。

与此同时，老板又提出了一些新的要求，比如能不能做到一些实时数据和历史数据的对比计算，加一些同比环比的计算，由于它是两套存储，两套业务的代码，很难做到数据口径上的对齐，因此做实时、离线的对比，很难能拿到一个精准的结果。这个时候有两种解决方案，一是让老板放弃需求，这通常不太可能。另一种方式是探索一种全新的解决办法，用一套统一的技术架构。

总结下来当前业界在流批一体方面的探索有两套典型的架构，第一个是 Lambda 架构，它存在需要两套割裂的计算存储等问题。另一种是基于 Flink 加实时的存储，比如用 Kafka 或 Paimon，把它叫 Kappa 架构，也可以叫 Lakehouse 架构。Lakehouse 架构虽然用一套架构，但通过增量计算的方式实现数据的实时更新，也存在一些问题，比如流计算的资源要常驻，回溯的成本高而且比较低效。对历史的数据做回刷，代码由于流计算和批计算的编程模型不一样，很难做到代码的的复用，使用这两种方案都会有一些各种各样的问题。综合以上现状，我们过去一年在 Flink 上做了一些流批一体的思考和探索，并取得了一些成果。

三、Materialized Table 构建流批一体 ETL

2.1 Materialized Table：声明式 ETL 统一流批作业

在介绍 Materialized Table 之前，我们先介绍一些背景信息。Apache Flink 是一个流批一体的计算引擎，从内核层面（包括 Runtime 和算子层面）到 Flink SQL 的 API 层面，Flink 一直致力于构建流批一体的计算引擎。然而，对于最上层的用户来说并没有做到真正的流批一体，本质上还是因为流计算和批计算编程模型不一样，流计算面向无限流编程，是增量计算的思路。批计算更多的面是向分区或表粒度，是全量数据的计算，因此写 SQL 代码的时候，更多的是面向分区粒度编程，这个时候代码层面很难做到复用。因此对用户来说，Flink SQL 并没有真正做到流批一体。基于这些现状和问题，在流批一体存储和 Flink 计算引擎在流批一体方面的能力完善之后，面向业务层面，我们抽象出一套流批一体的 API，叫做 Materialized Table。Materialized Table 可以理解为面向用户层面、业务层面做的流批一体的抽象实体。一个 Materialized Table SQL 语句可以把它分为三个部分，第一部分是 Create Materialized Table， 也就是关键字，说明这是一个物化表。第二部分是 Freshness 指定数据新鲜度，就是数据预期达到什么样的新鲜度。第三部分是 As Select 部分，指定 Query 的逻辑，数据生产的逻辑。用户只需要声明几行 SQL，Flink 引擎会根据用户指定的 Freshness 自动选择流批的执行模式，它是一个流或者一个批作业，然后自动刷新 Materialized Table 的数据。Materialized Table 想做的事情就是让用户不再去感知什么是流，什么是批，然后学流计算的、批计算的概念。Materialized Table 把这些东西屏蔽，让用户只关注于最核心的业务价值的部分，把其他的事情交给引擎。

2.2 从命令式 ETL 到声明式 ETL

从刚才的介绍看，使用 Materialized Table 构建新一代流批一体 ETL，相比于传统的数仓里面或者离线调度场景里面 ETL 有什么区别，首先看在传统数仓里面的 ETL，写一个完整的 ETL 作业，大概分这么几个部分，第一个部分是创建一张表，第二部分需要声明数据的分区也就是数据的新鲜度，需要的是过去一天的还是过去一小时的数据。第三部分是声明业务价值的部分，就是业务逻辑部分，涉及到表的一些小文件管理等操作。然后配置一个调度的作业，周期性的实现数据的刷新，达到想要的新鲜度。使用 Materialized Table 之后，只需要声明一套 SQL，也就是数据新鲜度和业务逻辑，把其他的事情都交给引擎，Materialized Table 相当于可以让你更专注于业务价值部分，实现一套 API 统一流和批。

从技术实现层面看，Materialized Table 相当于提供了一个用户层面的语法糖，在底层实现上相当于把原来的类似于命令式编程的过程交给引擎来完成。对用户来说可以说是从命令式的 ETL 到声明式的 ETL，从作业的视角切换为表的视角、数据的视角，只需要关注表和数据，从而提供数据库级别的体验。

2.3 一行 SQL 回刷历史数据

创建完表不意味着数据开发结束，后期可能会涉及数据需求的迭代，比如增加一些字段，或者删除一些字段，或者修改字段的计算逻辑。这个时候对于原来离线的数仓玩法，就需要重新写一个作业，更新业务逻辑，指定数据分区，然后再平台上通过调度去跑数据。使用 Materialized Table 后可以帮你一键完成这件事，只需要执行一条 ALTER MATERIALIZED TABLE customer_orders REFRESH PARTITION(ds=‘20241125’)完成指定数据分区的数据刷新。

2.4 一行 SQL 切换数据新鲜度

有时我们会面临着数据新鲜度调整的需求。比如一个典型的场景-电商场景，在平常的时候我们可能对数据新鲜度要求并不高，比如天级或小时级。在大促的时候，比如双 11 或 618，对数据新鲜度要求会比较高，比如分钟级或秒级。那这个时候怎么办呢？如果是传统的 Lambda 架构，相当于要新建一个实时的路，通过 Flink 加 Kafka 实现秒级数据更新，这就面临着重新开发、运维流作业等各种成本。使用 Materialized Table 可以通过一键完成数据新鲜度的修改，自动调整数据新鲜度，从而达到业务时效性，不用再重新搭建一套新的链路。以上是关于 Materialized Table 的一些面向用户层面最核心的 API。

2.5 技术架构

在介绍了 Materialized Table 的用户 API 之后，我们从开发者的视角来看看如何在公司内部实现和使用 Materialized Table。接下来，我们将从技术架构的角度梳理整体原理以及需要做的工作。

2.5.1 整体技术架构

Flink Materialized Table 的技术架构分为如下几个关键部分：

（1）Client：

• 用于提交与 Materialized Table 相关的 SQL 语句。

• 用户通过 Client 提交创建和管理 Materialized Table 的 SQL 命令。

（2）SQL Gateway：

• 一个常驻的服务，负责管理 Materialized Table 的全生命周期。

• 包括表的创建、元数据的管理、后台数据刷新作业的执行和监控等。

（3）Workflow：

• 用于处理周期性的批调度作业。

• 根据 Materialized Table 指定的数据新鲜度（如一天或一小时），定时执行数据刷新操作。

（4）CatalogStore：

• 用于存储和管理 Catalog 元数据。

• 与 Catalog 组件协同工作，确保元数据的持久化和一致性。

（5）Catalog：

• 核心组件，用于持久化 Materialized Table 的所有元数据，包括 Schema、定义的 Query、新鲜度以及对应的后台作业信息。

• 确保所有元数据能够被正确记录和管理。

（6）Flink Cluster：

• 执行数据刷新作业。

• 负责实际的数据处理和计算任务。

2.5.2 技术架构概述

整个技术架构基于 Flink 现有的能力，通过“搭积木”的方式将各个分散的组件拼接起来，形成一个更加完整的系统从而实现流批一体的能力。具体来说，这个架构并没有引入新的组件，而是通过整合现有 Flink 组件来实现所需功能。

2.5.3 需要对接的新组件

为了在公司内部使用 Materialized Table，开发者需要重点对接两个新的组件：

（1）Catalog：

• 作用：持久化 Materialized Table 的所有元数据，包括 Schema、定义的 Query、新鲜度以及对应的后台作业信息。

• 对接步骤：选择合适的 Catalog 实现（如具备流批一体存储的 Paimon Catalog）。Catalog 支持存储 Materialized Table 相关的元数据。

（2）Workflow 调度器：

• 作用：管理和执行周期性的批调度作业，根据 Materialized Table 指定的数据新鲜度进行定时刷新。

• 对接步骤：选择合适的 Workflow 调度工具（如 Airflow、DolphinScheduler 等）。配置 Workflow 以支持 Materialized Table 的刷新需求。实现定时任务的创建和管理，确保数据刷新的及时性和准确性。

2.6 Catalog API 对接

第一个是 Catalog API 的对接。在 Flink SQL 中提出了一个新的表类型，叫 CatalogMaterializedTable，它是和现有的 Flink 的 CatalogTable 平行的一个概念实体，但是它相比 CatalogTable多一些元数据，除 Schema 之外，多了 Definition Query，Freshness，后台作业元数据等。我们需要 Catalog API 对应的方法：createTable、getTable、alterTable，和 dropTable 支持这个新的表类型，就是实现对 CatalogMaterializedTable 对象的各种 CRUD 的操作。还有一个重要的方法叫 getFactory，因为 CatalogMaterializedTable 最终需要有后台的作业帮他完成数据的刷新，因此要编译它的 Defintion Query，然后提交作业执行，在编译的时候需要拿到它的 DynamicTableSourceFactory和 DynamicTableSinkFactory，这就需要通过 getFactory 方法获取。相当于 Materialized Table 需要 Catalog 具备存储的能力，就是 Catalog Connector 化。以上就是 Catalog 对接的时候需要完成的几个核心方法实现。在 Flink 社区里面我们已经完成和 Paimon 的集成，Paimon Catalog 已经支持 Materialized Table。假如你有自定义的 Catalog，想支持 Materialized Table，需要对应的实现这几个 Catalog 方法。

2.7 Pluggable Workflow 对接

第二部分是和 Workflow 的对接。用户在创建CatalogMaterializedTable 时候指定的新鲜度，后台可能对应的是一个流作业，也可能是一个批作业。假如他是一个批作业，需要依赖一个工作流调度器创建对应的 Workflow，然后通过它的周期性调度完成数据刷新，这个时候需要一个插件和 Workflow 调度器对接，让 SQL Gateway 和对应的工作流调度器相互通信，把这些东西接上。基于上述需求，我们在 FLIP-448 里面抽象出一个插件化的 WorkflowScheduler接口。WorkflowScheduler 接口有三个方法： createRefreshWorkflow，modifyRefreshWorkflow 和deleteRefreshWorkflow。对 Materialized Table 做一些 CRUD 的操作的时候，需要通过 WorkflowScheduler 接口和具体的工作流调度器通信，完成对应的操作。比如使用 Airflow 就需要实现一个AirflowWorkflowScheduler，使用 DolphinScheduler 就需要实现一个 DolphinSchedulerWorkflowScheduler，或者其他公司自研的工作流调度器就需要对应的实现。

如果想把 Materialized Table 用起来，从开发者的角度，需要核心对接的 API 就是以上两个部分。当前社区的进展是我们在 Flink 1.20 版本已经把CatalogMaterializedTable 和 Paimon 完成了对接，Paimon 的湖存储已经具备这个能力。和 WorkflowScheduler 对接，我们第一步会完成和 DolphinScheduler 的集成，这个会在 Flink 2.0 中完成，让 Materialized Table 能端到端的玩转起来，在 2.0 里面实现。当然我们在 2.0 里面也会做更多的事情，比如和 Yarn/K8S 的集成，让 Materialized Table 的作业能提交到 Yarn/K8S 上 YAML 开发室的对接，当前 1.20 只完成 MVP 版本的功能。

2.8 传统 Lambda 架构 Materialized Table 流批⼀体架构对比

传统 Lambda 架构的问题：

• 高成本：需要维护两套独立的系统（批处理和流处理），增加了硬件和运维成本。

• 低效率：数据处理逻辑在批处理和流处理中需要分别实现，代码复用性差，开发和维护效率低下。

• 易出错：批处理和流处理的数据处理逻辑可能存在差异，导致数据口径不一致。

Materialized Table 的优势：

• 低成本：通过一套存储和一套计算引擎，减少了硬件和运维成本。

• 高效率：提供统一的 API，用户只需编写一次 SQL 语句即可满足流批处理需求，提高了开发和维护效率。

• 正确性：统一的处理逻辑确保了数据的一致性和准确性。

Flink 的不同计算模式：

• 批计算：适用于全量数据处理。

• 流计算：适用于实时数据处理。

• 增量计算：正在探索中的计算模式，在分钟和小时级新鲜度场景可以带来的成本优化。

由于 Materialized Table 具备不同的计算能力，结合用户指定的 Freshness 以及成本优化器，会自动的选择一个最优的执行模式。这意味着 Materialized Table 能达到用户期望的成本和新鲜度的平衡，这也是用户层面最关心的事情。

Materialized Table 不仅解决了传统 Lambda 架构的痛点，还为用户和开发者提供了更高效、灵活和经济的解决方案，能够在满足业务需求的同时，提升整体的数据处理能力和用户体验。](https://ucc.alicdn.com/gfbp4bwpctdbo/developer-article1652183/20250214/e4ddbc0dcebe4747a664091e7df88da5.avif)

如今，数据湖（Data Lake）和湖仓一体（Lakehouse）的概念，以及像 Paimon 这样的技术栈，正变得越来越流行。理想的解决方案是以开放的数据湖格式作为核心，这种做法允许将多样化的数据源导入到统一的数据存储中。由于采用了开放格式，因此能够支持多种计算引擎，包括流处理、批处理、OLAP 分析甚至是 AI 模型训练等任务。通过这种方式，可以对数据执行多维度的分析与处理，并最终将结果应用于人工智能应用、商业智能报告或其他数据分析场景中，从而帮助企业更好地挖掘其数据资产的价值。

虽然这一架构看起来非常吸引人，并且很多组织已经在尝试实施，但在实际操作过程中可能会遇到一系列挑战。

接下来我们可以从一个案例看一下：假设 Steven 是某公司的数据工程师，上班之后接收到老板的钉钉消息——“能不能统计下平台昨天的一个热销商品，或者一些各品类的 GMV？”，熟悉业务的 Steven 选择 Iceberg 作为它的数据湖存储，然后把数据从 MySQL 里面通过 DataX 导进来，导完之后用 Spark 做 Batch ETL ，然后通过 QuickBI 搭建一个很漂亮的大屏，老板看到大屏之后很满意。然后提出一个新的需求。

就是实现报表每天更新，接到需求后 Steven 在现有的批链路上加一个调度器，比如用 Airflow，实现每天调度批作业，实现了报表的每天更新。

于是老板又期望报表每天都实时更新，Steven 对 Iceberg 比较熟，知道 Iceberg 当前主要面向离线场景，在实时更新上可能有一些困难。Spark 或者 Spark SQL 很难做到一个秒级或分钟级的实时计算，在现有的统一的湖仓架构的链路上无法满足，这时候不得不搭建一套新的实时链路。比如用 Flink 加 Kafka 典型的组合完成一套实时的链路，然后重新搭建一套实时的 BI 大屏。这时候相当于引入实时和离线的两套链路，本质上就是一个 Lambda 的架构。

与此同时，老板又提出了一些新的要求，比如能不能做到一些实时数据和历史数据的对比计算，加一些同比环比的计算，由于它是两套存储，两套业务的代码，很难做到数据口径上的对齐，因此做实时、离线的对比，很难能拿到一个精准的结果。这个时候有两种解决方案，一是让老板放弃需求，这通常不太可能。另一种方式是探索一种全新的解决办法，用一套统一的技术架构。

总结下来当前业界在流批一体方面的探索有两套典型的架构，第一个是 Lambda 架构，它存在需要两套割裂的计算存储等问题。另一种是基于 Flink 加实时的存储，比如用 Kafka 或 Paimon，把它叫 Kappa 架构，也可以叫 Lakehouse 架构。Lakehouse 架构虽然用一套架构，但通过增量计算的方式实现数据的实时更新，也存在一些问题，比如流计算的资源要常驻，回溯的成本高而且比较低效。对历史的数据做回刷，代码由于流计算和批计算的编程模型不一样，很难做到代码的的复用，使用这两种方案都会有一些各种各样的问题。综合以上现状，我们过去一年在 Flink 上做了一些流批一体的思考和探索，并取得了一些成果。

三、Materialized Table 构建流批一体 ETL

2.1 Materialized Table：声明式 ETL 统一流批作业

在介绍 Materialized Table 之前，我们先介绍一些背景信息。Apache Flink 是一个流批一体的计算引擎，从内核层面（包括 Runtime 和算子层面）到 Flink SQL 的 API 层面，Flink 一直致力于构建流批一体的计算引擎。然而，对于最上层的用户来说并没有做到真正的流批一体，本质上还是因为流计算和批计算编程模型不一样，流计算面向无限流编程，是增量计算的思路。批计算更多的面是向分区或表粒度，是全量数据的计算，因此写 SQL 代码的时候，更多的是面向分区粒度编程，这个时候代码层面很难做到复用。因此对用户来说，Flink SQL 并没有真正做到流批一体。基于这些现状和问题，在流批一体存储和 Flink 计算引擎在流批一体方面的能力完善之后，面向业务层面，我们抽象出一套流批一体的 API，叫做 Materialized Table。Materialized Table 可以理解为面向用户层面、业务层面做的流批一体的抽象实体。一个 Materialized Table SQL 语句可以把它分为三个部分，第一部分是 Create Materialized Table， 也就是关键字，说明这是一个物化表。第二部分是 Freshness 指定数据新鲜度，就是数据预期达到什么样的新鲜度。第三部分是 As Select 部分，指定 Query 的逻辑，数据生产的逻辑。用户只需要声明几行 SQL，Flink 引擎会根据用户指定的 Freshness 自动选择流批的执行模式，它是一个流或者一个批作业，然后自动刷新 Materialized Table 的数据。Materialized Table 想做的事情就是让用户不再去感知什么是流，什么是批，然后学流计算的、批计算的概念。Materialized Table 把这些东西屏蔽，让用户只关注于最核心的业务价值的部分，把其他的事情交给引擎。

2.2 从命令式 ETL 到声明式 ETL

从刚才的介绍看，使用 Materialized Table 构建新一代流批一体 ETL，相比于传统的数仓里面或者离线调度场景里面 ETL 有什么区别，首先看在传统数仓里面的 ETL，写一个完整的 ETL 作业，大概分这么几个部分，第一个部分是创建一张表，第二部分需要声明数据的分区也就是数据的新鲜度，需要的是过去一天的还是过去一小时的数据。第三部分是声明业务价值的部分，就是业务逻辑部分，涉及到表的一些小文件管理等操作。然后配置一个调度的作业，周期性的实现数据的刷新，达到想要的新鲜度。使用 Materialized Table 之后，只需要声明一套 SQL，也就是数据新鲜度和业务逻辑，把其他的事情都交给引擎，Materialized Table 相当于可以让你更专注于业务价值部分，实现一套 API 统一流和批。

从技术实现层面看，Materialized Table 相当于提供了一个用户层面的语法糖，在底层实现上相当于把原来的类似于命令式编程的过程交给引擎来完成。对用户来说可以说是从命令式的 ETL 到声明式的 ETL，从作业的视角切换为表的视角、数据的视角，只需要关注表和数据，从而提供数据库级别的体验。

2.3 一行 SQL 回刷历史数据

创建完表不意味着数据开发结束，后期可能会涉及数据需求的迭代，比如增加一些字段，或者删除一些字段，或者修改字段的计算逻辑。这个时候对于原来离线的数仓玩法，就需要重新写一个作业，更新业务逻辑，指定数据分区，然后再平台上通过调度去跑数据。使用 Materialized Table 后可以帮你一键完成这件事，只需要执行一条 ALTER MATERIALIZED TABLE customer_orders REFRESH PARTITION(ds=‘20241125’)完成指定数据分区的数据刷新。

2.4 一行 SQL 切换数据新鲜度

有时我们会面临着数据新鲜度调整的需求。比如一个典型的场景-电商场景，在平常的时候我们可能对数据新鲜度要求并不高，比如天级或小时级。在大促的时候，比如双 11 或 618，对数据新鲜度要求会比较高，比如分钟级或秒级。那这个时候怎么办呢？如果是传统的 Lambda 架构，相当于要新建一个实时的路，通过 Flink 加 Kafka 实现秒级数据更新，这就面临着重新开发、运维流作业等各种成本。使用 Materialized Table 可以通过一键完成数据新鲜度的修改，自动调整数据新鲜度，从而达到业务时效性，不用再重新搭建一套新的链路。以上是关于 Materialized Table 的一些面向用户层面最核心的 API。

2.5 技术架构

在介绍了 Materialized Table 的用户 API 之后，我们从开发者的视角来看看如何在公司内部实现和使用 Materialized Table。接下来，我们将从技术架构的角度梳理整体原理以及需要做的工作。

2.5.1 整体技术架构

Flink Materialized Table 的技术架构分为如下几个关键部分：

（1）Client：

• 用于提交与 Materialized Table 相关的 SQL 语句。

• 用户通过 Client 提交创建和管理 Materialized Table 的 SQL 命令。

（2）SQL Gateway：

• 一个常驻的服务，负责管理 Materialized Table 的全生命周期。

• 包括表的创建、元数据的管理、后台数据刷新作业的执行和监控等。

（3）Workflow：

• 用于处理周期性的批调度作业。

• 根据 Materialized Table 指定的数据新鲜度（如一天或一小时），定时执行数据刷新操作。

（4）CatalogStore：

• 用于存储和管理 Catalog 元数据。

• 与 Catalog 组件协同工作，确保元数据的持久化和一致性。

（5）Catalog：

• 核心组件，用于持久化 Materialized Table 的所有元数据，包括 Schema、定义的 Query、新鲜度以及对应的后台作业信息。

• 确保所有元数据能够被正确记录和管理。

（6）Flink Cluster：

• 执行数据刷新作业。

• 负责实际的数据处理和计算任务。

2.5.2 技术架构概述

整个技术架构基于 Flink 现有的能力，通过“搭积木”的方式将各个分散的组件拼接起来，形成一个更加完整的系统从而实现流批一体的能力。具体来说，这个架构并没有引入新的组件，而是通过整合现有 Flink 组件来实现所需功能。

2.5.3 需要对接的新组件

为了在公司内部使用 Materialized Table，开发者需要重点对接两个新的组件：

（1）Catalog：

• 作用：持久化 Materialized Table 的所有元数据，包括 Schema、定义的 Query、新鲜度以及对应的后台作业信息。

• 对接步骤：选择合适的 Catalog 实现（如具备流批一体存储的 Paimon Catalog）。Catalog 支持存储 Materialized Table 相关的元数据。

（2）Workflow 调度器：

• 作用：管理和执行周期性的批调度作业，根据 Materialized Table 指定的数据新鲜度进行定时刷新。

• 对接步骤：选择合适的 Workflow 调度工具（如 Airflow、DolphinScheduler 等）。配置 Workflow 以支持 Materialized Table 的刷新需求。实现定时任务的创建和管理，确保数据刷新的及时性和准确性。

2.6 Catalog API 对接

第一个是 Catalog API 的对接。在 Flink SQL 中提出了一个新的表类型，叫 CatalogMaterializedTable，它是和现有的 Flink 的 CatalogTable 平行的一个概念实体，但是它相比 CatalogTable多一些元数据，除 Schema 之外，多了 Definition Query，Freshness，后台作业元数据等。我们需要 Catalog API 对应的方法：createTable、getTable、alterTable，和 dropTable 支持这个新的表类型，就是实现对 CatalogMaterializedTable 对象的各种 CRUD 的操作。还有一个重要的方法叫 getFactory，因为 CatalogMaterializedTable 最终需要有后台的作业帮他完成数据的刷新，因此要编译它的 Defintion Query，然后提交作业执行，在编译的时候需要拿到它的 DynamicTableSourceFactory和 DynamicTableSinkFactory，这就需要通过 getFactory 方法获取。相当于 Materialized Table 需要 Catalog 具备存储的能力，就是 Catalog Connector 化。以上就是 Catalog 对接的时候需要完成的几个核心方法实现。在 Flink 社区里面我们已经完成和 Paimon 的集成，Paimon Catalog 已经支持 Materialized Table。假如你有自定义的 Catalog，想支持 Materialized Table，需要对应的实现这几个 Catalog 方法。

2.7 Pluggable Workflow 对接

第二部分是和 Workflow 的对接。用户在创建CatalogMaterializedTable 时候指定的新鲜度，后台可能对应的是一个流作业，也可能是一个批作业。假如他是一个批作业，需要依赖一个工作流调度器创建对应的 Workflow，然后通过它的周期性调度完成数据刷新，这个时候需要一个插件和 Workflow 调度器对接，让 SQL Gateway 和对应的工作流调度器相互通信，把这些东西接上。基于上述需求，我们在 FLIP-448 里面抽象出一个插件化的 WorkflowScheduler接口。WorkflowScheduler 接口有三个方法： createRefreshWorkflow，modifyRefreshWorkflow 和deleteRefreshWorkflow。对 Materialized Table 做一些 CRUD 的操作的时候，需要通过 WorkflowScheduler 接口和具体的工作流调度器通信，完成对应的操作。比如使用 Airflow 就需要实现一个AirflowWorkflowScheduler，使用 DolphinScheduler 就需要实现一个 DolphinSchedulerWorkflowScheduler，或者其他公司自研的工作流调度器就需要对应的实现。

如果想把 Materialized Table 用起来，从开发者的角度，需要核心对接的 API 就是以上两个部分。当前社区的进展是我们在 Flink 1.20 版本已经把CatalogMaterializedTable 和 Paimon 完成了对接，Paimon 的湖存储已经具备这个能力。和 WorkflowScheduler 对接，我们第一步会完成和 DolphinScheduler 的集成，这个会在 Flink 2.0 中完成，让 Materialized Table 能端到端的玩转起来，在 2.0 里面实现。当然我们在 2.0 里面也会做更多的事情，比如和 Yarn/K8S 的集成，让 Materialized Table 的作业能提交到 Yarn/K8S 上 YAML 开发室的对接，当前 1.20 只完成 MVP 版本的功能。

2.8 传统 Lambda 架构 Materialized Table 流批⼀体架构对比

传统 Lambda 架构的问题：

• 高成本：需要维护两套独立的系统（批处理和流处理），增加了硬件和运维成本。

• 低效率：数据处理逻辑在批处理和流处理中需要分别实现，代码复用性差，开发和维护效率低下。

• 易出错：批处理和流处理的数据处理逻辑可能存在差异，导致数据口径不一致。

Materialized Table 的优势：

• 低成本：通过一套存储和一套计算引擎，减少了硬件和运维成本。

• 高效率：提供统一的 API，用户只需编写一次 SQL 语句即可满足流批处理需求，提高了开发和维护效率。

• 正确性：统一的处理逻辑确保了数据的一致性和准确性。

Flink 的不同计算模式：

• 批计算：适用于全量数据处理。

• 流计算：适用于实时数据处理。

• 增量计算：正在探索中的计算模式，在分钟和小时级新鲜度场景可以带来的成本优化。

由于 Materialized Table 具备不同的计算能力，结合用户指定的 Freshness 以及成本优化器，会自动的选择一个最优的执行模式。这意味着 Materialized Table 能达到用户期望的成本和新鲜度的平衡，这也是用户层面最关心的事情。

Materialized Table 不仅解决了传统 Lambda 架构的痛点，还为用户和开发者提供了更高效、灵活和经济的解决方案，能够在满足业务需求的同时，提升整体的数据处理能力和用户体验。

03. Demo

接下来播放一段 Demo，演示基于 Flink Materialized Table 加 Paimon 构建流批一体 ETL。

点击此处跳转DEMO

以上就是本次的分享内容，谢谢大家～

更多内容

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版现开启活动：新用户复制点击下方链接或者扫描二维码即可 0 元免费试用 Flink + Paimon实时计算 Flink 版（3000CU*小时，3 个月内）了解活动详情：https://free.aliyun.com/?utm_content=g_1000395379&productCode=sc