目录
一:启动脚本解析
在 /bin/start-seatunnel-flink.sh
#!/bin/bash
function usage() {
echo "Usage: start-seatunnel-flink.sh [options]"
echo " options:"
echo " --config, -c FILE_PATH Config file"
echo " --variable, -i PROP=VALUE Variable substitution, such as -i city=beijing, or -i date=20190318"
echo " --check, -t Check config"
echo " --help, -h Show this help message"
}
if [[ "$@" = *--help ]] || [[ "$@" = *-h ]] || [[ $# -le 1 ]]; then
usage
exit 0
fi
is_exist() {
if [ -z $1 ]; then
usage
exit -1
fi
}
PARAMS=""
while (( "$#" )); do
case "$1" in
-c|--config)
CONFIG_FILE=$2
is_exist ${CONFIG_FILE}
shift 2
;;
-i|--variable)
variable=$2
is_exist ${variable}
java_property_value="-D${variable}"
variables_substitution="${java_property_value} ${variables_substitution}"
shift 2
;;
*) # preserve positional arguments
PARAMS="$PARAMS $1"
shift
;;
esac
done
if [ -z ${CONFIG_FILE} ]; then
echo "Error: The following option is required: [-c | --config]"
usage
exit -1
fi
# set positional arguments in their proper place
eval set -- "$PARAMS"
BIN_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
APP_DIR=$(dirname ${BIN_DIR})
CONF_DIR=${APP_DIR}/config
PLUGINS_DIR=${APP_DIR}/lib
DEFAULT_CONFIG=${CONF_DIR}/application.conf
CONFIG_FILE=${CONFIG_FILE:-$DEFAULT_CONFIG}
assemblyJarName=$(find ${PLUGINS_DIR} -name seatunnel-core-flink*.jar)
if [ -f "${CONF_DIR}/seatunnel-env.sh" ]; then
source ${CONF_DIR}/seatunnel-env.sh
fi
string_trim() {
echo $1 | awk '{$1=$1;print}'
}
export JVM_ARGS=$(string_trim "${variables_substitution}")
exec ${FLINK_HOME}/bin/flink run \
${PARAMS} \
-c org.apache.seatunnel.SeatunnelFlink \
${assemblyJarName} --config ${CONFIG_FILE}
org.apache.seatunnel.SeatunnelFlink 这个类就是主入口
二:源码解析
-
入口
public class SeatunnelFlink {
public static void main(String[] args) throws Exception {
FlinkCommandArgs flinkArgs = CommandLineUtils.parseFlinkArgs(args);
Seatunnel.run(flinkArgs);
}
}
FlinkCommandArgs中进行命令行参数解析
public static FlinkCommandArgs parseFlinkArgs(String[] args) {
FlinkCommandArgs flinkCommandArgs = new FlinkCommandArgs();
JCommander.newBuilder()
.addObject(flinkCommandArgs)
.build()
.parse(args);
return flinkCommandArgs;
}
进入到Seatunnel.run(flinkArgs);
public static <T extends CommandArgs> void run(T commandArgs) {
if (!Common.setDeployMode(commandArgs.getDeployMode().getName())) {
throw new IllegalArgumentException(
String.format("Deploy mode: %s is Illegal", commandArgs.getDeployMode()));
}
try {
Command<T> command = CommandFactory.createCommand(commandArgs);
command.execute(commandArgs);
} catch (ConfigRuntimeException e) {
showConfigError(e);
throw e;
} catch (Exception e) {
showFatalError(e);
throw e;
}
}
进入到CommandFactory.createCommand(commandArgs);,根据不同的类型选择Command,我们看的是flinkCommand。
public static <T extends CommandArgs> Command<T> createCommand(T commandArgs) {
switch (commandArgs.getEngineType()) {
case FLINK:
return (Command<T>) new FlinkCommandBuilder().buildCommand((FlinkCommandArgs) commandArgs);
case SPARK:
return (Command<T>) new SparkCommandBuilder().buildCommand((SparkCommandArgs) commandArgs);
default:
throw new RuntimeException(String.format("engine type: %s is not supported", commandArgs.getEngineType()));
}
}
进入到 buildCommand,根据是否检查config进入到不同的实现类
public Command<FlinkCommandArgs> buildCommand(FlinkCommandArgs commandArgs) {
return commandArgs.isCheckConfig() ? new FlinkConfValidateCommand() : new FlinkTaskExecuteCommand();
}
FlinkConfValidateCommand和FlinkTaskExecuteCommand两个类都实现了Command类。并且都只有一个execute()方法
public class FlinkConfValidateCommand implements Command<FlinkCommandArgs>
public class FlinkTaskExecuteCommand extends BaseTaskExecuteCommand<FlinkCommandArgs, FlinkEnvironment>
在SeaTunnel.run(flinkArgs)进入 command.execute(commandArgs);
我们先看FlinkTaskExecuteCommand 类中的execute方法
2.execute()核心方法
public void execute(FlinkCommandArgs flinkCommandArgs) {
//flink
EngineType engine = flinkCommandArgs.getEngineType();
// --config
String configFile = flinkCommandArgs.getConfigFile();
//将String变成Config类
Config config = new ConfigBuilder<>(configFile, engine).getConfig();
//解析执行上下文
ExecutionContext<FlinkEnvironment> executionContext = new ExecutionContext<>(config, engine);
//解析 sources模块
List<BaseSource<FlinkEnvironment>> sources = executionContext.getSources();
//解析 tansform模块
List<BaseTransform<FlinkEnvironment>> transforms = executionContext.getTransforms();
//解析 sink模块
List<BaseSink<FlinkEnvironment>> sinks = executionContext.getSinks();
baseCheckConfig(sinks, transforms, sinks);
showAsciiLogo();
try (Execution<BaseSource<FlinkEnvironment>,
BaseTransform<FlinkEnvironment>,
BaseSink<FlinkEnvironment>,
FlinkEnvironment> execution = new ExecutionFactory<>(executionContext).createExecution()) {
//准备
prepare(executionContext.getEnvironment(), sources, transforms, sinks);
//启动
execution.start(sources, transforms, sinks);
//关闭
close(sources, transforms, sinks);
} catch (Exception e) {
throw new RuntimeException("Execute Flink task error", e);
}
}