王炸组合：Dolphinscheudler 3.1.*搭配SeaT unnel2.3.*高效完成异构数据数据集成

file

概述

本篇主要介绍如何通过Dolphinscheduler海豚调度搭配Seatunnel完成异构数据源之间的数据同步功能，这个在大数据流批一体数仓建设的过程中是一个非常好的解决方案，稳定高效，只要用上了你肯定爱不释手。

环境准备

dolphinscheduler集群 >= 3.1.5
dolphinscheduler3.1.5版本源码
Seatunnel集群 >= 2.3.3

没有安装好以上准备环境的童鞋，请先参考我的另外两篇文章完成基础环境搭建基于Seatunnel最新2.3.5版本分布式集群安装部署指南(小白版)及dolphinscheduler分布式集群部署指南(小白版)再回到章节继续。

配置文件修改

这里说明一下，通过海豚调度配置的Seatunnel数据同步任务最后都会被分配到DS集群的某个Worker组或者某个worker节点上进行执行，所以你要保证你的DS集群的目标worker节点上也安装了Seatunnel服务。这很重要，因为实际dolphisncheduler中定义的seatunnel任务实例到最后都是需要调用worker节点上的seatunnel服务在本地执行seatunnel的任务启动命令来完成任务提交和运行。

Dolphinscheduler的配置文件修改

因为我们需要使用seatunnel完成数据集成，所以我们需要在dolphinscheduler的系统环境变量中将我们的Seatunnel的安装目录进行配置。

找到你的dolphinscheduler主节点的安装目录下的$DOLPHINSCHEDULER_HOME/bin/env/dolphinscheduler_env.sh。

设置SEATUNNEL_HOME的访问目录，将SEATUNNEL_HOME设置为你自己的SeaTunnel安装目录。

export SEATUNNEL_HOME=${SEATUNNEL_HOME:-/opt/software/seatunnel-2.3.5}

然后保存重启Dolphinscheduler集群即可完成配置修改同步到所有的api-server、master-server及worker-server节点。

Dolphinscheduler部分源码修改

为什么要修改Dolphinscheduler的源码？因为我这里使用的Seatunnel的版本是2.3.5，使用的引擎不是Seatunnel的默认引擎，用的是Spark引擎， Spark我用的版本是2.4.5，所有我最后在命令执行的命令如下：

$SEATUNNEL_HOME/bin/start-seatunnel-spark-2-connector-v2.sh --master local[4] --deploy-mode client --config /opt/software/seatunnel-2.3.5/config/app-config/v2.batch.config.template

如果我用的是Spark3.X的版本，我执行命令如下：

$SEATUNNEL_HOME/bin/start-seatunnel-spark-3-connector-v2.sh --master local[4] --deploy-mode client --config /opt/software/seatunnel-2.3.5/config/app-config/v2.batch.config.template

然而在Dolphinscheduler3.1.5版本的Seatunnel任务插件中，存在一些问题没办法兼容，首先是前端，这里引擎只支持Spark和Flink，没有针对具体的版本进行兼容，没办法自由的选择使用Spark2、Spark3还是FIink13、Flink15。 file

其次就是后端的代码。 file

找到EngineEnum类，修改一下代码如下： file

public enum EngineEnum {// FLINK("${SEATUNNEL_HOME}/bin/start-seatunnel-flink.sh"),// SPARK("${SEATUNNEL_HOME}/bin/start-seatunnel-spark.sh");FLINK("${SEATUNNEL_HOME}/bin/start-seatunnel-flink-13-connector-v2.sh"),FLINK15("${SEATUNNEL_HOME}/bin/start-seatunnel-flink-15-connector-v2.sh"),SPARK("${SEATUNNEL_HOME}/bin/start-seatunnel-spark-2-connector-v2.sh"),SPARK3("${SEATUNNEL_HOME}/bin/start-seatunnel-spark-3-connector-v2.sh");private String command;EngineEnum(String command) {this.command = command;}public String getCommand() {return command;}
}

这样修改完毕之后就OK了，然后编译打包dolphinscheduler的源码。

更新Dolphinscheduler集群中的SeaTunnel任务插件

项目编译打包完成之后，找到dolphinscheduler-task-seatunnel工程下的target目录下的dolphinscheduler-task-seatunnel-3.1.5.jar包，上传到你的dolphinscheduler集群的主节点。

file

然后将主节点上DS安装目录下的api-server/libs、master-server/libs、worker-server/libs、alert-server/libs目录(其实这里可以只替换woker-server/libs目录)下的dolphinscheduler-task-seatunnel-3.1.5.jar重命名为dolphinscheduler-task-seatunnel-3.1.5.jar.20240606(带上日期方便知道变更时间)。

file

然后将我们编译的dolphinscheduler-task-seatunnel-3.1.5.jar拷贝到这几个目录(api-server/libs、master-server/libs、worker-server/libs、alert-server/libs目录,确认一下是不是所有目录下都有这个dolphinscheduler-task-seatunnel-3.1.5.jar，没有的目录就直接略过)下。

然后使用主节点上的分发脚本，将api-server/libs、master-server/libs、worker-server/libs、alert-server/libs的修改同步到其他的DS节点上，分发完成之后，检查一下分发是否成功。

最后就是重启我们的DS集群，通过以上步骤我们就完成了Dolphisncheduler中SeaTunnel插件的升级适配。

测试验证

我们通过dolphinscheduler的工作流定义页面定义一个Seatunnel数据同步的任务，完成Oracle数据库表采集到MySQL数据库的任务，下面我们来操作。

关于seatunnel任务配置脚本文件，官网的文档介绍如下：

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/category/source-v2
Transform: https://seatunnel.incubator.apache.org/zh-CN/docs/category/transform-v2
Sink: https://seatunnel.incubator.apache.org/zh-CN/docs/category/sink-v2/

Source输入源配置定义说明

这里我们的输入原始Oracle, 所以直接从Source中查找Oracle相关的配置如何定义，官网给我们提供了不少任务示例，：

简单任务示例

# Defining the runtime environment
env {parallelism = 4job.mode = "BATCH"
}
source{Jdbc {url = "jdbc:oracle:thin:@datasource01:1523:xe"driver = "oracle.jdbc.OracleDriver"user = "root"password = "123456"query = "SELECT * FROM TEST_TABLE"}
}transform {}sink {Console {}
}

按分区列并行任务示例

并行读取你配置的分片字段和分片数据，如果你想读取整个表，可以这样做

env {parallelism = 4job.mode = "BATCH"
}
source {Jdbc {url = "jdbc:oracle:thin:@datasource01:1523:xe"driver = "oracle.jdbc.OracleDriver"connection_check_timeout_sec = 100user = "root"password = "123456"# 根据需要定义查询逻辑query = "SELECT * FROM TEST_TABLE"# 设置并行分片读取字段partition_column = "ID"# 分区切片数量partition_num = 10properties {database.oracle.jdbc.timezoneAsRegion = "false"}}
}
sink {Console {}
}

按主键或唯一索引并行任务示例

配置table_path会开启自动分割，可以配置split.*来调整分割策略

env {parallelism = 4job.mode = "BATCH"
}
source {Jdbc {url = "jdbc:oracle:thin:@datasource01:1523:xe"driver = "oracle.jdbc.OracleDriver"connection_check_timeout_sec = 100user = "root"password = "123456"table_path = "DA.SCHEMA1.TABLE1"query = "select * from SCHEMA1.TABLE1"split.size = 10000}
}sink {Console {}
}

并行上下限任务示例

指定查询的上下限内的数据效率更高按照你配置的上下限来读取你的数据源效率更高

source {Jdbc {url = "jdbc:oracle:thin:@datasource01:1523:xe"driver = "oracle.jdbc.OracleDriver"connection_check_timeout_sec = 100user = "root"password = "123456"# Define query logic as requiredquery = "SELECT * FROM TEST_TABLE"partition_column = "ID"# Read start boundarypartition_lower_bound = 1# Read end boundarypartition_upper_bound = 500partition_num = 10}
}

多表读取任务示例

配置table_list会开启自动分割，可以通过配置split.来调整分割策略*

env {job.mode = "BATCH"parallelism = 4
}
source {Jdbc {url = "jdbc:oracle:thin:@datasource01:1523:xe"driver = "oracle.jdbc.OracleDriver"connection_check_timeout_sec = 100user = "root"password = "123456""table_list"=[{"table_path"="XE.TEST.USER_INFO"},{"table_path"="XE.TEST.YOURTABLENAME"}]#where_condition= "where id > 100"split.size = 10000#split.even-distribution.factor.upper-bound = 100#split.even-distribution.factor.lower-bound = 0.05#split.sample-sharding.threshold = 1000#split.inverse-sampling.rate = 1000}
}sink {Console {}
}

Sink输出源配置定义说明

简单任务示例

本示例定义了一个SeaTunnel同步任务，通过FakeSource自动生成数据并发送到JDBC Sink。FakeSource一共生成16行数据（row.num=16），每行有两个字段name（string类型）和age（int类型）。最终的目标表为test_table，表中同样会有16行数据。运行此作业之前，你需要在mysql中创建数据库test和表test_table。如果你还没有安装和部署SeaTunnel，你需要按照安装SeaTunnel中的说明安装并部署SeaTunnel。然后按照快速开始使用SeaTunnel引擎中的说明运行此作业。

env {parallelism = 1job.mode = "BATCH"
}source {FakeSource {parallelism = 1result_table_name = "fake"row.num = 16schema = {fields {name = "string"age = "int"}}}
}transform {}sink {jdbc {url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"driver = "com.mysql.cj.jdbc.Driver"user = "root"password = "123456"query = "insert into test_table(name,age) values(?,?)"}
}

生成输出SQL任务示例

本示例无需编写复杂的sql语句，您可以配置输出端数据库名称表名称来自动为您生成添加语句

sink {jdbc {url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"driver = "com.mysql.cj.jdbc.Driver"user = "root"password = "123456"# 根据数据库表名自动生成sql语句generate_sink_sql = truedatabase = testtable = test_table}
}

精确任务示例

对于需要精确写入场景，我们保证精确一次。

sink {jdbc {url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"driver = "com.mysql.cj.jdbc.Driver"max_retries = 0user = "root"password = "123456"query = "insert into test_table(name,age) values(?,?)"is_exactly_once = "true"xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"}
}

CDC(变更数据捕获)事件

我们也支持CDC变更数据在这种情况下，您需要配置数据库，表和primary_keys。

sink {jdbc {url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"driver = "com.mysql.cj.jdbc.Driver"user = "root"password = "123456" generate_sink_sql = true# You need to configure both database and tabledatabase = testtable = sink_tableprimary_keys = ["id","name"]field_ide = UPPERCASEschema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"data_save_mode="APPEND_DATA"}
}

完整测试脚本配置文件

下面给出本示例中完整的配置文件示例

env {parallelism = 4job.mode = "BATCH"
}
source{Jdbc {url = "jdbc:oracle:thin:@192.168.11.101:15210:YLAPP"driver = "oracle.jdbc.OracleDriver"user = "appuser001"password = "appuser001"query = "SELECT * FROM YL_APP.MET_COM_ICDOPERATION_LS"}
}transform {}sink {jdbc {url = "jdbc:mysql://192.168.10.210:13306/yl-app-new?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"driver = "com.mysql.cj.jdbc.Driver"user = "appuser001"password = "appuser001" generate_sink_sql = "true"database = "hive"table = "met_com_icdoperation_ls"schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"data_save_mode="APPEND_DATA"}
}

file

将上述脚本中的数据库配置信息修改成你的数据连接配置，然后将脚本覆盖到上图脚本输入中，保存工作流，上线之后启动工作流。

file

到对应数据库验证

原来的Oracle数据库表

file

同步之后的MySQL数据库表

file

任务成功了，数据也成功同步过来了， OK,测试通过！大家接下来可以在这个Demo的基础上进行更多的扩展和挖掘，实战的多了，你对于Dolphinscheduler和Seatunnel的架构和原理的理解就会越来越深入了，慢慢你也可以通过扩展源码来升级和拓展这些优秀开源框架的功能了。创作不易，如果我的文章对你有帮助欢迎点赞，收藏，送你一朵小红花~

原文链接：https://blog.csdn.net/qq_41865652/article/details/140971419