记第一次跟踪seatunnel的任务运行过程四——getJobConfigParser().parse()的动作

发布于:2025-03-17 ⋅ 阅读:(20) ⋅ 点赞:(0)

前绪

记第一次跟踪seatunnel的任务运行过程三——解析配置的具体方法getLogicalDag

正文

书接上文

ImmutablePair<List<Action>, Set<URL>> immutablePair = getJobConfigParser().parse(null);

在前一篇文章中说到getLogicDag()方法的第一行(如上),执行了一个解析方法获得了action和jar包资源对,这一步就是由parse()方法来执行的。本片文章就是对parse()方法的源码探索。

任务配置文件

先回顾一下执行seatunnel.sh时命令中‘ --config ’指定的任务配置文件的结构:

env{}
source{}
transform{}
sink{}

主要有以上四部分

获得任务配置解析器

getJobConfigParser()方法也org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment中的一个受保护的方法,在这个方法中直接返回了一个MultipleTableJobConfigParser(多表任务配置解析器)对象。这也与Seatunnel官网宣传的支持耽搁文件中同时同步多张表的数据相一致。
所以对parse()方法的调用,实际就是由MultipleTableJobConfigParser的parse方法的承担执行的。
在getJobConfigParser()方法方法中,有对‘从保存的检查点开始任务’的支持。但是从执行的代码路径上来看,创建ClientJobExecutionEnvironment对象的时候,其中isStartWithSavePoint参数给的是false,所以并没有启用。
确实我们的探索是从一个全新的任务开始并不是从一个save point开始。

填充jobClient对象(解析env)

this.fillJobConfigAndCommonJars();

获取‘ --config ’指定的配置文件中的env部分的内容:

  • 提取任务模式(jobModel),是stream流式的任务,还是batch批任务;
  • 提取任务命名;
  • 提取其他env部分的配置;
  • 提取env部分指定的第三方jar包的uri,并且封装为commonPluginJars(公共插件jar包);

解析source、tranform、sink

List<? extends Config> sourceConfigs = TypesafeConfigUtils.getConfigList(seaTunnelJobConfig, "source", Collections.emptyList());
List<? extends Config> transformConfigs = TypesafeConfigUtils.getConfigList(seaTunnelJobConfig, "transform", Collections.emptyList());
List<? extends Config> sinkConfigs =  TypesafeConfigUtils.getConfigList(seaTunnelJobConfig, "sink", Collections.emptyList());

解析source、transform、sink部分的内容,获取各部分的配置信息。

解析connector

List<URL> sourceConnectorJars = getConnectorJarList(sourceConfigs, PluginType.SOURCE);
List<URL> transformConnectorJars = getConnectorJarList(transformConfigs, PluginType.TRANSFORM);
List<URL> sinkConnectorJars = getConnectorJarList(sinkConfigs, PluginType.SINK);

获得source、transform、sink三部分的使用的connector信息。

换用seatunnel自己的classLoader

ClassLoader parentClassLoader = Thread.currentThread().getContextClassLoader();

将当前使用中的classLoader暂存。

List<URL> sourceJars = Stream.of(sourceConnectorJars, transformConnectorJars)
                        .flatMap(Collection::stream)
                        .distinct()
                        .collect(Collectors.toList());
ClassLoader sourceAndTransformClassLoader = getClassLoader(classLoaderService, parentClassLoader, sourceJars);
ClassLoader sinkClassLoader = getClassLoader(classLoaderService, parentClassLoader, sinkConnectorJars);

将source和transform的connector jar包资源合并去重,然后分别创建“source与transform共用的classLoader”以及“sink使用的classLoader”。

Thread.currentThread().setContextClassLoader(sourceAndTransformClassLoader);

先使用“source与transform共用的classLoader”进行处理

检查配置文件是否可有构建成DAG

ConfigParserUtil.checkGraph(sourceConfigs, transformConfigs, sinkConfigs);

检查source、sink、transform(如有)能否组成一个DAG。

checkGraph方法中

if (CollectionUtils.isEmpty(sources) || CollectionUtils.isEmpty(sinks)) {
    throw new JobDefineCheckException("Source And Sink can not be null");
}

如果缺少source或sink直接报错,source和sink作为执行任务的必备元素,不可或缺。

if (isSimpleGraph(sources, transforms, sinks)) {
    checkSimpleGraph(sources, transforms, sinks);
       return;
    }
checkComplexGraph(sources, transforms, sinks);

如果是简单的图(graph)则执行简单查询,执行完直接返回,否则就执行复杂检查。

是否是简单图:isSimpleGraph(sources, transforms, sinks)
sources.size() == 1 && sinks.size() == 1 && (CollectionUtils.isEmpty(transforms) || transforms.size() == 1)

如果source下只有一个内容,并且sink下也只有一个内容,并且没有transform或只有一个transform,则认为是简单图。
这也是最常用的两种配置模式

  1. 单source+单sink
  2. 单source+单transform+单sink
执行简单检查:checkSimpleGraph(sources, transforms, sinks)
ReadonlyConfig source = ReadonlyConfig.fromConfig(sources.get(0));
ReadonlyConfig sink = ReadonlyConfig.fromConfig(sinks.get(0));
if (transforms.size() == 0) {
    checkEdge(source, sink);
} else {
    ReadonlyConfig transform = ReadonlyConfig.fromConfig(transforms.get(0));
    checkEdge(source, transform);
    checkEdge(transform, sink);
}
  • 如果没有transform,就检查source的plugin_output(之前版本是result_table_name)和sink的plugin_input(之前版本是source_table_name)是否匹配。
  • 如果有transform,就检查source的plugin_output(之前版本是result_table_name)和transform的plugin_input(之前版本是source_table_name)是否匹配;以及就检查transform的plugin_output(之前版本是result_table_name)和sink的plugin_input(之前版本是source_table_name)是否匹配。
    在此过程中,若配置文件中没有配置plugin_input和plugin_output,则使用默认的名称default-identifier代替。

解析source、transform、sink配置内容

for (int configIndex = 0; configIndex < sourceConfigs.size(); configIndex++) {
                Config sourceConfig = sourceConfigs.get(configIndex);
                Tuple2<String, List<Tuple2<CatalogTable, Action>>> tuple2 =
                        parseSource(configIndex, sourceConfig, sourceAndTransformClassLoader);
                tableWithActionMap.put(tuple2._1(), tuple2._2());
            }

            log.info("start generating all transforms.");
            parseTransforms(transformConfigs, sourceAndTransformClassLoader, tableWithActionMap);

            Thread.currentThread().setContextClassLoader(sinkClassLoader);
            log.info("start generating all sinks.");
            List<Action> sinkActions = new ArrayList<>();
            for (int configIndex = 0; configIndex < sinkConfigs.size(); configIndex++) {
                Config sinkConfig = sinkConfigs.get(configIndex);
                sinkActions.addAll(
                        parseSink(configIndex, sinkConfig, sinkClassLoader, tableWithActionMap));
            }
            Set<URL> factoryUrls = getUsedFactoryUrls(sinkActions);
            return new ImmutablePair<>(sinkActions, factoryUrls);

逐步解析source、transform、sink的配置内容,最终组装成action返回。
从MultipleTableJobConfigParser的类名刻制,这是支持多表配置文件解析的,所以在解析source、transform、sink时都有循环便利的代码存在。

结束

 Thread.currentThread().setContextClassLoader(parentClassLoader);
            if (classLoaderService != null) {
                classLoaderService.releaseClassLoader(Long.parseLong(jobConfig.getJobContext().getJobId()), sourceJars);
                classLoaderService.releaseClassLoader( Long.parseLong(jobConfig.getJobContext().getJobId()), sinkConnectorJars);
            }

换回默认的类加载器,并释放“source与transform共用的classLoader”以及“sink使用的classLoader”资源。