结项报告完整版 | Apache SeaTunnel 支持 metalake 开发

过去两周，我们对开源之夏活动中表现优异的开发者们进行了简单的采访，初步粗略地了解了一下他们的开发过程和心得体会。今天，我们将通过同学们的完整结项报告，深入了解项目的开发技术细节，希望能够帮助大家更好地了解 Apache SeaTunnel 项目的最新进展。

接下来是关于 Apache SeaTunnel 支持 metalake 开发这一项目的完整报告：

一、项目背景

目前，Apache SeaTunnel 的任务配置中，数据源的用户名和密码等敏感信息直接写死在任务脚本中，这种方式 存在以下问题：

1. 安全隐患：敏感信息暴露在脚本中，易导致数据源信息泄漏。

2. 维护困难： 数据源配置信息发生变更时，需手动修改所有相关任务脚本，效率低下且易出错。

为解决上述问题，本项目旨在通过集成 metalake，实现数据源信息的集中存储和管理。通过数据源 ID 映射机 制，用户可方便地更新和管理数据源配置。本项目的目标是支持主流数据目录 Apache Gravitino，并通过预留 接口，方便扩展支持其他第三方数据目录服务。

Apache Gravitino 获取数据源配置信息的 REST API 示例见于：https://gravitino.apache.org/docs/0.9.0- incubating/api/rest/load-catalog

代码仓库见于： https://github.com/apache/seatunnel

1. 完成 metalake 配置信息适配将 metalake 配置信息配置在 seatunnel-env 中，任务启动后加载到任务配置脚本的 env 中。

2. 1.1 任务启动时读取seatunnel-env中的配置项。1.2 将配置集成到任务脚本的env中，确保任务能够正确加载metalake配置。

3. 完成 source 和 sink 的数据源配置信息改造读取env中是否开启 metalake 标识，在 source 和 sink 中增加sourceId作为查询 metalake 的唯一标识，获取数据 源信息并替换 source/sink 配置项中的占位符。

4. 2.1 在 source 和 sink 配置中增加 sourceId 配置项。2.2 支持 source/sink 配置项中的占位符替换，通过 sourceId 动态获取数据源信息。

5. 插件方式支持 metalake 并集成 Apache Gravitino 定义 metalake 接口，支持根据唯一 ID 查询数据源配置信息，并实现 Apache Gravitino 数据源信息转换为 SeaTunnel 配置项占位符的功能。3.1 定义 metalake 实现接口，提供数据源查询功能。3.2 支持 Apache Gravitino 集成，参考 Gravitino REST API 文档。3.3 支持扩展性，通过实现接口可支持其他数据目录，如 UnityCatalog 或 DataHub。3.4 确保向后兼容，不影响存量任务的正常运行。

二、方案描述

1. 完成 metalake 配置信息适配

2. 1.1 任务启动时读取seatunnel-env中配置项

• 实现目标：在任务启动时，从seatunnel-env.sh或者任务配置文件中读取 metalake 相关的配置。

• 实现方法：在seatunnel-env.sh文件中定义 metalake 配置项，例如：

METALAKE_ENABLED=trueMETALAKE_TYPE=gravitinoMETALAKE_URL=http://localhost:8090/api/metalakes/metalake_name/catalogs /...

或者在任务配置文件中的 env 中配置

env{metalake_enabled = truemetalake_type = "gravitino"metalake_url ="http://localhost:8090/api/metalakes/metalake_name/catalogs/" }

1.2 将配置集成到 env 中

• 实现目标：将读取的 metalake 配置集成到任务的 env 中。

• 实现方法：

• 若用户在任务配置文件中配置 env，那么自然无需集成。

• 若在 seatunnel-env.sh 脚本中配置，也可通过 System.getEnv()获得，无需集成到 env 中

2. 完成 source 和 sink 的数据源配置信息改造

3. 2.1 source/sink 增加 sourceId 配置项

• 实现目标：为 source 和 sink 添加 sourceId 字段，用于标识 metalake 中的数据源。

• 实现方法：

• 在任务脚本中指定 sourceId 即可。

• 任务脚本示例：

source {type = "mysql"sourceId = "mysql_datasource_001"url = "jdbc:mysql://localhost:3306/db"...}

2.2 支持 source/sink 的配置项占位符替换

• 实现目标：通过 metalake 动态获取数据源信息，并替换配置中的占位符。

• 实现方法：

• 在配置解析阶段，检查 sourceId 和 metalakeEnabled。

• 如果启用 metalake 且 sourceId 存在，则用户可将 username 和 password 等字段并设为占位符，然 后通过 metalake 接口查询数据源信息并占位符替换。

• 步骤：

• 定义占位符格式，例如 ${key}。

• 通过 REST API 查询数据源信息。

• 

• 替换配置中的占位符。

• 代码示例：

3. 插件方式支持 metalake 并支持 Apache Gravitino 集成

4. 3.1 定义 metalake 实现接口

5. 实现目标：定义一个通用接口，用于与 metalake 交互。

6. 实现方法：

7. 定义MetalakeClient接口，包含查询数据源信息的方法。

8. 接口定义：

9. 

10. 3.2 支持 Apache Gravitino 集成

• 实现目标：实现与 Apache Gravitino 的集成，通过 REST API 获取数据源信息。

• 实现方法：

• 创建GravitinoClient类，实现MetalakeClient接口。

• 使用 HTTP 客户端发送请求到 Gravitino API，并解析响应。

• 代码示例：

• 

• 3.3 支持可扩展

• 实现目标：通过插件化设计，支持其他 metalake 实现。

• 实现方法：

• 使用工厂方法，根据 metalakeType 选择合适的 client。

• 代码示例：

• 

• 3.4 不影响存量任务，向后兼容

• 实现目标：确保新功能不破坏现有任务。

• 实现方法：

• 将 metalakeEnabled 设为可选配置，默认值为 false。

• 仅在 metalakeEnabled=true 且 sourceId 存在时触发 metalake 逻辑。

• 代码示例：

• 

三、时间规划

四、项目进度

已完成工作

已完成项目所需功能的开发与测试，并经过修改后，已经合并了 PR。

遇到的问题与解决方案

在编写代码时，我遇到的问题不多，并且要感谢 liugddx 老师的指导，在老师的指导下，我遇到的问题基本迎刃 而解。

还有一个问题就是该项目的 test case 较多，测试时间较长，并且合并 PR 前要通过所有的 test case。然后由于网 络等原因，这些 test case 不是很稳定，有时需要多次重试才能通过，这很考验我的耐心。

测试用例

设计了一个简单的任务配置脚本，并在 source 中使用了 metalake。

并为此测试用例构建了相应的 MySQL 数据库和 Gravitino。在 sink 中使用了 Assert connector，保证得到正确的结 果。该集成测试的 test case 代码也已上传 github，并且通过了测试。

后续工作安排

后续可以考虑集成更多的 metalake 类型，使得该功能不局限于 Gravitino。