commandLine

commandLine.hasOption(CliFrontendOptions.HELP))

{HelpFormatter

HelpFormatter();formatter.setLeftPadding(4);formatter.setWidth(80);formatter.printHelp(

Create

pipelinePipelineExecution.ExecutionInfo

result

createExecutor(commandLine).run();//

Print

resultprintExecutionInfo(result);}这里首先是初始化了一下选项,这里用到了Apache

Commons

这个工具类,可以很方便的处理命令行参数大概的步骤有3步1.定义阶段

定义要解析的命令选项,对应的每个选项就是一个Option类,Options类是Option类的一个集合2.解析阶段

通过CommandLineParser的parser方法将main方法的args参数解析,获得一个CommandLine对象3.查询阶段

就是具体使用解析后的结果,可以通过hasOption来判断是否有该选项,getOptionValue来获取选项对应的值

具体可以参考我的另外一系列文章,有详细介绍这个工具的用法超强命令行解析工具

Apache

解析了入参后就判断入参args是否是空或者是否包含-h或者–help这个选项,如果是的话就打印一下帮助信息

最后在打印一下结果信息

PipelineExecution.ExecutionInfo

result

createExecutor(commandLine).run();static

CliExecutor

IllegalArgumentException(Missing

pipeline

Paths.get(unparsedArgs.get(0));if

(!Files.exists(pipelineDefPath))

{throw

FileNotFoundException(String.format(Cannot

find

getGlobalConfig(commandLine);//

Load

getFlinkHome(commandLine);Configuration

flinkConfig

FlinkEnvironmentUtils.loadFlinkConfiguration(flinkHome);//

Additional

Arrays.stream(Optional.ofNullable(commandLine.getOptionValues(CliFrontendOptions.JAR)).orElse(new

String[0])).map(Paths::get).collect(Collectors.toList());//

Build

CliExecutor(pipelineDefPath,flinkConfig,globalPipelineConfig,commandLine.hasOption(CliFrontendOptions.USE_MINI_CLUSTER),additionalJars);}可以看到最后是构建了一个CliExecutor类,并执行了它的run方法.

选项类

CLI这个类库,代码比较简单这里主要细看一下各个选项都有什么作用

package

org.apache.flink.cdc.cli;import

import

org.apache.commons.cli.Options;/**

Command

Option.builder().longOpt(flink-home).hasArg().desc(Path

Flink

Option.builder(h).longOpt(help).desc(Display

help

Option.builder().longOpt(global-config).hasArg().desc(Path

***

Option.builder().longOpt(jar).hasArgs().desc(JARs

submitted

Option.builder().longOpt(use-mini-cluster).hasArg(false).desc(Use

Flink

Options().addOption(HELP).addOption(JAR).addOption(FLINK_HOME).addOption(GLOBAL_CONFIG).addOption(USE_MINI_CLUSTER);}

–flink-home

指定flink-home的地址,有了这个参数我们就可以不使用系统环境自带的FLINK_HOME,可以使用指定的flink版本

–global-config

flink-cdc.yaml文件,这里面的参数很少,我看只有配置一个并发度,其他的配置没看到,这块有感兴趣的老铁可以再仔细看看

–jar

使用mini-cluster模式启动,mini-cluster相当于就是本地local模式启动,会用多个现成模拟JobManager,TaskManager,ResourceManager,Dispatcher等组件,一般用于测试

--help

com.ververica.cdc.cli.parser.PipelineDefinitionParser;

import

com.ververica.cdc.cli.parser.YamlPipelineDefinitionParser;

import

com.ververica.cdc.cli.utils.FlinkEnvironmentUtils;

import

com.ververica.cdc.common.annotation.VisibleForTesting;

import

com.ververica.cdc.common.configuration.Configuration;

import

com.ververica.cdc.composer.PipelineComposer;

import

com.ververica.cdc.composer.PipelineExecution;

import

com.ververica.cdc.composer.definition.PipelineDef;import

java.nio.file.Path;

pipelineDefPath;this.flinkConfig

flinkConfig;this.globalPipelineConfig

globalPipelineConfig;this.useMiniCluster

useMiniCluster;this.additionalJars

PipelineExecution.ExecutionInfo

run()

YamlPipelineDefinitionParser();PipelineDef

pipelineDef

pipelineDefinitionParser.parse(pipelineDefPath,

Create

composer.compose(pipelineDef);//

Execute

FlinkEnvironmentUtils.createComposer(useMiniCluster,

flinkConfig,

composer;}VisibleForTestingvoid

composer)

composer;}VisibleForTestingpublic

Configuration

flinkConfig;}VisibleForTestingpublic

Configuration

globalPipelineConfig;}VisibleForTestingpublic

ListPath

方法首先是构建了一个yaml解析器用于解析yaml配置文件然后调用parser

获得一个PipelineDef类,这相当与将yaml配置文件转换成了一个配置实体Bean,方便之后操作接下来获取到PipelineComposer对象,然后调用compose

方法传入刚刚的配置实体BeanPiplineDef对象,就获得了一个PiplineExecution对象最后调用execute方法启动任务(这个方法底层就是调用了flink

StreamExecutionEnvironment.executeAsync方法)

配置文件解析类

################################################################################

Description:

################################################################################

source:type:

table.create.properties.light_schema_change:

truetable.create.properties.replication_num:

1route:-

doris_sync.table_\.*sink-table:

Sync

2这个类的主要目标就是要将这个yaml文件解析成一个实体类PipelineDef方便之后的操作代码解释就直接写到注释中了

package

com.ververica.cdc.cli.parser;import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;

import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;

import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;import

com.ververica.cdc.common.configuration.Configuration;

import

com.ververica.cdc.composer.definition.PipelineDef;

import

com.ververica.cdc.composer.definition.RouteDef;

import

com.ververica.cdc.composer.definition.SinkDef;

import

com.ververica.cdc.composer.definition.SourceDef;import

java.nio.file.Path;

com.ververica.cdc.common.utils.Preconditions.checkNotNull;/**

Parser

解析的核心工具方法,可以获取yaml文件中的值,或者将其中的值转换成java实体类private

final

mapper.readTree(pipelineDefPath.toFile());//

Source

toSourceDef(checkNotNull(root.get(SOURCE_KEY),

source

toSinkDef(checkNotNull(root.get(SINK_KEY),

sink

这里是路由配置,是个数组,而且是个可选项,所以这里优雅的使用了Optional对root.get(ROUTE_KEY)

然后调用ifPresent方法来判断,如果参数存在的时候才会执行的逻辑,就是遍历数组然后加到

routeDefs

ArrayList();Optional.ofNullable(root.get(ROUTE_KEY)).ifPresent(node

node.forEach(route

routeDefs.add(toRouteDef(route))));//

Pipeline

参数,是可选项,这个如果不指定,配置就是用的flink-cdc中的配置Configuration

userPipelineConfig

toPipelineConfig(root.get(PIPELINE_KEY));//

Merge

用户配置,这的addAll实际上就是HashMap的putAll,新值会把旧值覆盖,所以用户的配置优先级大于全局配置Configuration

pipelineConfig

Configuration();pipelineConfig.addAll(globalPipelineConfig);pipelineConfig.addAll(userPipelineConfig);//

new

mapper.convertValue(sourceNode,

new

checkNotNull(sourceMap.remove(TYPE_KEY),

将type

Configuration.fromMap(sourceMap));}private

SinkDef

checkNotNull(sinkMap.remove(TYPE_KEY),Missing

required

sinkMap.remove(NAME_KEY);return

new

Configuration.fromMap(sinkMap));}private

RouteDef

checkNotNull(routeNode.get(ROUTE_SOURCE_TABLE_KEY),Missing

required

configuration,ROUTE_SOURCE_TABLE_KEY).asText();

从JsonNode

checkNotNull(routeNode.get(ROUTE_SINK_TABLE_KEY),Missing

required

configuration,ROUTE_SINK_TABLE_KEY).asText();String

description

Optional.ofNullable(routeNode.get(ROUTE_DESCRIPTION_KEY)).map(JsonNode::asText).orElse(null);return

new

mapper.convertValue(pipelineConfigNode,

new

Configuration.fromMap(pipelineConfigMap);}

配置信息工具类

com.ververica.cdc.cli.utils;import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;

import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

import

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;import

com.ververica.cdc.common.configuration.Configuration;import

import

FileNotFoundException(String.format(Cannot

find

mapper.readValue(configPath.toFile(),

new

Configuration.fromMap(configMap);}

catch

IllegalStateException(String.format(Failed

load

YAMLFactory());这行代码主要是用了Jackson库中的两个核心类,ObjectMapper和YAMLFactoryObjectMapper

Jackson

库中用于序列化将对象转换为字节流或其他格式和反序列化将字节流或其他格式转换为对象的核心类。

它提供了各种方法来处理

JSON、YAML

mapper.readValue(configPath.toFile(),

new

{});这行的意思就是传入yaml配置文件,容纳后将其转换成一个Map类型,kv都是String因为这个类的主要用途是解析global-conf的,也就是conf目录下的flink-cdc.yaml,这个文件仅只有kv类型的,所以要转换成mapflink-cdc.yaml

Parallelism

EVOLVE这里再简单看一下mapper的readValue方法Jackson

ObjectMapper的readValue方法主要用途就是将配置文件转换成java实体,主要可以三个重载

public

这个一般不常用最后这行就是将一个map转换成Configuration对象

return

Configuration.fromMap(configMap);这里的Configuration就是将HashMap做了一个封装,方便操作

FLink环境工具类

com.ververica.cdc.cli.utils;import

com.ververica.cdc.common.configuration.Configuration;

import

com.ververica.cdc.composer.flink.FlinkPipelineComposer;import

java.nio.file.Path;

flinkHome.resolve(FLINK_CONF_DIR).resolve(FLINK_CONF_FILENAME);return

ConfigurationUtils.loadMapFormattedConfig(flinkConfPath);}public

static

FlinkPipelineComposer.ofMiniCluster();}return

FlinkPipelineComposer.ofRemoteCluster(org.apache.flink.configuration.Configuration.fromMap(flinkConfig.toMap()),additionalJars);}

public

flinkHome.resolve(FLINK_CONF_DIR).resolve(FLINK_CONF_FILENAME);return

ConfigurationUtils.loadMapFormattedConfig(flinkConfPath);}这个方法的主要目的就是通过找到FLINK_HOME/conf/flink-conf.yaml文件,然后将这个文件转换成一个Configuration对象,转换的方法在上一节中介绍过了这里还用到了Path

resolve

FlinkPipelineComposer.ofMiniCluster();}return

FlinkPipelineComposer.ofRemoteCluster(org.apache.flink.configuration.Configuration.fromMap(flinkConfig.toMap()),additionalJars);}这个是通过一些参数来初始化Composer,Composer就是将用户的任务翻译成一个Pipeline作业的核心类这里首先是判断了一下是否使用miniCluster,如果是的话就用minicluster

来指定,具体在上文中介绍过.如果不是那么就用remoteCluster,这里就不多介绍了,之后的文章会介绍

上面几个类写的比较多,这里做一个总结,简单的来总结一下这个模块flink-cdc-cli

模块的主要作用1.解析任务配置yaml文件,转换成一个PipelineDef任务实体类2.通过FLINK_HOME获取flink的相关配置信息,然后构建出一个PipelineComposer

对象3.调用composer的comoose方法,传入任务实体类获取PipelineExecution任务执行对象,然后启动任务

解析配置文件生成任务实体类,然后启动任务

解析yaml文件3.加深了对Optional类判断null值的印象,之后对于null值判断有个一个更优雅的写法4.对flink-cdc-cli模块有了个全面的认识,但是具体还有些细节需要需要深入到其他模块再去了解

[mini-cluster介绍]

https://www.cnblogs.com/wangwei0721/p/14052016.html[Jackson

使用]

https://www.cnblogs.com/del88/p/13098678.html