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

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

代码环境搭建

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

• 在本地使用 IDE（如 IntelliJ IDEA）克隆 fork 后的代码，并配置 Java 以及 Maven 仓库。

• 编译拉取的代码，并确保代码格式化：

• 编译构建项目：

• 解压后的代码目录：

• 启动服务端：

学习与实践 SeaTunnel 的工作流程

• 官方提供的 SeaTunnel 工作流图：

• 以 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    }  }}

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

代码阅读

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

• 通过命令行提交完会调用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 配置容器信息。

个人心得

总结

本文由 科技 提供发布支持！