解析 Apache SeaTunnel 的任务运行过程

示例代码运行

感兴趣的朋友可以先看看这篇文章《记本地第一次运行示例项目》：https://blog.csdn.net/u011924665/article/details/143373017

基于分支

2.3.5-Release

跟踪过程

入口

Apache SeaTunnel提供的官方Example基于SeaTunnel Engine引擎的示例作为入口：org.apache.seatunnel.example.engine.SeaTunnelEngineExample

构建SeaTunnel引擎的命令行运行对象

这里默认会使用seatunnel-engine-example子项目的resources下的example/fanke\_to\_console.conf这个任务配置文件，并且以本地方式运行。

首先解析任务配置文件，构建ClientCommandArgs对象。

 public static void main(String[] args)            throws FileNotFoundException, URISyntaxException, CommandException {        String configurePath = args.length > 0 ? args[0] : "/examples/fake_to_console.conf";        String configFile = getTestConfigFile(configurePath);        ClientCommandArgs clientCommandArgs = new ClientCommandArgs();        clientCommandArgs.setConfigFile(configFile);        clientCommandArgs.setCheckConfig(false);        clientCommandArgs.setJobName(Paths.get(configFile).getFileName().toString());        // Change Execution Mode to CLUSTER to use client mode, before do this, you should start        // SeaTunnelEngineServerExample        clientCommandArgs.setMasterType(MasterType.LOCAL);        SeaTunnel.run(clientCommandArgs.buildCommand());    }

最后一行中，ClientCommandArgs的buildCommand()方法会根据任务配置文件构建出一个Command对象。

默认配置文件构建出来的就是一个ClientExecuteCommand（org.apache.seatunnel.core.starter.seatunnel.command包下）对象，简而言之，这就是一个通过命令行执行的SeaTunnel引擎对象。

ClientExecuteCommand开始运行job

在seatunnel-engine-example示例代码中最后的SeaTunnel.run()方法就是调用ClientExecuteCommand的execute()方法开始，然后通过命令行的方式运行SeaTunnel的Job。

推荐以下步骤，对比源码一起阅读。

声明Job监控器对象的引用

JobMetricsRunner.JobMetricsSummary jobMetricsSummary = null;

整个Execute的核心内容都被包裹在一个try-catch-finally代码块中，此处声明一个Job监控器是为了在后续的Finally代码块中获取到任务执行结束（无论是成功还是失败）后的统计信息。

获取SeaTunnel的配置

SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig();

在seatunnel-engine-example中获取的是SeaTunnel任务配置，而在此处获取的是SeaTunnel的配置，生成配置的对象。

获取Haselcast的客户端的配置

ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();

与加载SeaTunnel的配置类似，加载Hazelcast的Client配置，生成配置对象。

本地方式运行

clusterName = creatRandomClusterName(             StringUtils.isNotEmpty(clusterName)                 ? clusterName : Constant.DEFAULT_SEATUNNEL_CLUSTER_NAME);

本地方式运行且没有设置集群名称，则会默认以SeaTunnel开头后面添加随机数，生成一个集群名称。

instance = createServerInLocal(clusterName, seaTunnelConfig);

然后创建一个本地的Hazelcast客户端，创建成功之后，记录下来Hazelcast的端口号，用于在执行SeaTunnel的Job时使用。

确认（记录）集群名称

将本次Job执行的集群名称记录到SeaTunnel和SeaTunnel Client的配置对象中。

创建SeaTunnel Engine客户端

engineClient = new SeaTunnelClient(clientConfig);

创建一个SeaTunnel Engine的客户端，基于Hazelcast创建的。

创建出Engine Client对象后就是一列的判断。看起来是通过启动命令中指定来实现了，推测与任务恢复、状态等信息查询有关，暂未深入探究，感兴趣的小伙伴 可以研究研究。

准备执行Job

Path configFile = FileUtils.getConfigPath(clientCommandArgs);checkConfigExist(configFile);JobConfig jobConfig = new JobConfig();

获取Job的配置文件，并对文件进行检查，然后创建Job配置对象。

jobExecutionEnv = engineClient    .createExecutionContext(configFile.toString(), jobConfig, seaTunnelConfig);

使用Job配置对象，基于通过前面创建的SeaTunnel Engine客户端创建一个Job运行环境对象jobExecutionEnv。

开始执行

ClientJobProxy clientJobProxy = jobExecutionEnv.execute();

通过调用jobExecutionEnv的execute()方法开始执行Jb，并返回一个Job代理对象：ClientJobProxy。

对execute()方法的跟踪见：记第一次跟踪seatunnel的任务运行过程二——ClientJobExecutionEnvironment的execture方法

然后检查一下当前Job运行是否为异步模式，若为异步模式且不是本地模式运行，则流程结束。

若为非异步模式或任务是本地方式（seatunnel-engine-example中默认的是本地模式）执行，则继续执行以下流程。

注册任务取消的钩子函数

打印监控信息

启动一个新的定时线程，用于打印监控信息。

JobMetricsRunner jobMetricsRunner = new JobMetricsRunner(engineClient, jobId);executorService = Executors.newSingleThreadScheduledExecutor(                                new ThreadFactoryBuilder()                                        .setNameFormat("job-metrics-runner-%d")                                        .setDaemon(true)                                        .build());executorService.scheduleAtFixedRate(        jobMetricsRunner,        0,        seaTunnelConfig.getEngineConfig().getPrintJobMetricsInfoInterval(),        TimeUnit.SECONDS);

这个线程打印的就是如下的监控信息：

等到Job执行结果

线程在此同步的等待Job的执行结果。

结尾

如果Job的结果的错误信息不为空或者Job的结果为失败，则抛出异常。

正常结束则打印出Job的统计信息，形式如下：

    ***********************************************               Job Statistic Information    ***********************************************    Start Time                : 2024-10-31 19:02:18    End Time                  : 2024-10-31 19:05:32    Total Time(s)             :                 194    Total Read Count          :              239202    Total Write Count         :              239202    Total Failed Count        :                   0    ***********************************************

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