中电信翼康工程师：我在 Apache SeaTunnel 社区的贡献之旅

SeaTunnel

于 2024-10-25 10:11:43 发布

阅读量916 大数据

本文链接：https://blog.csdn.net/weixin_54625990/article/details/143227476

贡献者Github ID：luckyLJY 文章整理：曾辉

Apache SeaTunnel 作为一款强大的数据同步和转换工具，凭借其部署易用性、容错机制、数据源支持、性能优势、功能丰富性以及活跃的社区支持，成为了数据工程师们不可或缺的利器。

因其具有的以下优良特性：

在数据源方面，支持多种传统数据库、以及现有流行的多数大数据存储；

数据同步方面，支持批流的数据同步方式，完美兼容离线同步、实时同步、全量同步、增量同步等实际应用场景；

其社区在技术更新上非常迅速，确保了中电信翼康能够及时获取技术支持与更新迭代。将其选型作为了数据集成主要技术。

SeaTunnel 不仅能够满足各种规模和类型的数据处理需求，为用户提供高效、稳定和灵活的数据处理解决方案。随着数据环境的不断演变和发展，SeaTunnel 将继续在数据同步和转换领域发挥领导作用，推动数据驱动的业务发展。

对于我们公司基于SeaTunnel的应用实践也可以参考这篇文章：https://mp.weixin.qq.com/s/SitGSc9iZYlH-nLpGR0AJw

在了解了 Apache SeaTunnel 的背景后，接下来我将分享如何快速参与其代码开发，完善功能。以自己首次贡献代码的经历为例，简述具体步骤。

代码环境搭建

首先，在 SeaTunnel 的官方 GitHub 仓库地址：https://github.com/apache/seatunnel 进行 fork 操作，将代码仓库复制到自己的 GitHub 账户中。

在本地使用 IDE（如 IntelliJ IDEA）克隆 fork 后的代码，并配置 Java 以及 Maven 仓库。
编译拉取的代码，并确保代码格式化：

编译构建项目：

对编译打包完成的 SeaTunnel 代码进行测试：
- 解压后的代码目录：

启动服务端：

提交测试案例：

测试成功的环境如下：

这一步非常重要，它确保你拥有一个正确的开发环境，以便后续代码开发的顺利进行。

学习与实践 SeaTunnel 的工作流程

官方提供的 SeaTunnel 工作流图：

通过该图我们可以看出，SeaTunnel 的工作流程包括：source（数据源）、transform（数据转换）、sink（数据目标），以及执行引擎。source 是数据的输入源，sink 是数据的输出目标，而 transform 则是将输入数据转换为目标数据的插件。SeaTunnel 支持多种执行引擎，包括其自带的 Zeta 引擎，以及 Flink 和 Spark。

以 MySQL 到 MySQL 数据同步为例，以下是一个实际案例的 conf 配置文件：

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  jdbc {
    url = "jdbc:mysql://localhost:3306/test_his"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    query = "select * from ods_jc_hzjbxxb;"
    properties {
      useSSL=false
      rewriteBatchedStatements=true
    }
  }
}

transform {
}

sink {
  jdbc {
    url = "jdbc:mysql://localhost:3306/api-server"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    query = """insert into ods_jc_hzjbxxb (...) values (...);"""
    properties {
      useSSL=false
      rewriteBatchedStatements=true
    }
  }
}

该配置实现了 MySQL 到 MySQL 的简单数据同步。请根据官方文档中的 connector-v2 示例，创建 JDBC 插件，并在 plugins/jdbc/lib 目录下放置 MySQL 连接器。然后重启 SeaTunnel 服务，将配置文件放在 config 目录下，并使用以下命令提交作业：

./bin/seatunnel.cmd --config ./config/mysql_jdbc.conf

通过日志可以验证数据同步是否成功，并在目标数据库中检查同步结果。

代码阅读

通过学习 SeaTunnel 的工作流程，主要调试 SeaTunnel 自带的Zeta引擎的执行过程，通过官方文档的阅读了解到 SeaTunnel 的作业运行模式有Local和集群模式，项目实际应用更多为集群模式，因此把调试集群的执行过程作为主要学习对象。

服务端执行顺序的代码调试：

通过 server.execute() 方法查看服务端的执行流程：

通过代码调试的手段了解客户端启动的执行顺序

通过命令行提交完会调用org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand#execute方法
调用org.apache.seatunnel.engine.client.job.ClientJobProxy#submitJob方法
调用服务端的org.apache.seatunnel.engine.server.CoordinatorService#submitJob

尝试开发

需求背景
我们在公司开发 OpenGauss CDC 时，发现当前的 JDBC Sink 插件不直接支持 OpenGauss 数据库，而是通过 PostgreSQL JDBC Sink 来写入。

实际使用中遇到了兼容性问题，因此我决定开发一个通过 OpenGauss 自带 JDBC 驱动进行连接的功能。
代码实现
通过阅读 JDBC 连接器的代码，我发现只需要在 seatunnel-connectors-v2/connector-jdbc 模块下实现 OpenGauss 的 Catalog 和 Dialect。
- org.apache.seatunnel.connectors.seatunnel.jdbc.catalog 包下创建 OpenGauss 的 CatalogFactory，因其兼容 PostgreSQL，可继承 Postgres 的实现。
- 在 org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect 包下实现 OpenGauss 的方言工厂，覆写 acceptsURL 方法以支持 OpenGauss 的连接。
e2e 测试代码编写
编写 e2e 测试需要继承 AbstractJdbcIT 类，并实现相关的测试案例，如 initContainer 初始化容器，以及 getJdbcCase 配置容器信息。