Flink源码学习（5）Flink编程的执行流程

ak918xp / 2024-04-21 / 原文

Flink上层的API很简单，编程套路较为固定

执行环境 ExecutionEnvironment

数据抽象 DataSet/DataStream

逻辑算子 Source Tramsform Sink

我们以Flink中提供的AdaptiveSchedulerITCase为例子

@Test
public void testGlobalFailoverCanRecoverState() throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(PARALLELISM);

    env.enableCheckpointing(20L, CheckpointingMode.EXACTLY_ONCE);
    final DataStreamSource<Integer> input = env.addSource(new SimpleSource());

    input.addSink(new DiscardingSink<>());

    env.execute();
}

Env->Datasource->Addsink->Execute

在flink应用程序中所有的操作都是StreamOperator，分为Source，Sink，Stream

内置优化：多个特定能被优化的operator会形成chain

三个类似的概念：

Function高阶算子的参数->形成Operator->底层的表示Tramsform

Flink内置入门程序

public class SocketWindowWordCount {

    public static void main(String[] args) throws Exception {

        // the host and the port to connect to
        final String hostname;
        final int port;
        try {
            final ParameterTool params = ParameterTool.fromArgs(args);
            hostname = params.has("hostname") ? params.get("hostname") : "localhost";
            port = params.getInt("port");
        } catch (Exception e) {
            System.err.println(
                    "No port specified. Please run 'SocketWindowWordCount "
                            + "--hostname <hostname> --port <port>', where hostname (localhost by default) "
                            + "and port is the address of the text server");
            System.err.println(
                    "To start a simple text server, run 'netcat -l <port>' and "
                            + "type the input text into the command line");
            return;
        }

        // get the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // get input data by connecting to the socket
        DataStream<String> text = env.socketTextStream(hostname, port, "\n");

        // parse the data, group it, window it, and aggregate the counts
        DataStream<WordWithCount> windowCounts =
                text.flatMap(
                                (FlatMapFunction<String, WordWithCount>)
                                        (value, out) -> {
                                            for (String word : value.split("\\s")) {
                                                out.collect(new WordWithCount(word, 1L));
                                            }
                                        },
                                Types.POJO(WordWithCount.class))
                        .keyBy(value -> value.word)
                        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                        .reduce((a, b) -> new WordWithCount(a.word, a.count + b.count))
                        .returns(WordWithCount.class);

        // print the results with a single thread, rather than in parallel
        windowCounts.print().setParallelism(1);

        env.execute("Socket Window WordCount");
    }

    // ------------------------------------------------------------------------

    /** Data type for words with count. */
    public static class WordWithCount {

        public String word;
        public long count;

        @SuppressWarnings("unused")
        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}

编写好之后打成jar包

mvn clean package -DskipTests

通过命令提交

flink命令脚本通过Java命令启动，在ClieFrontend类来启动jvm进程执行任务的构造和提交

Flink run xxx.jar class arg1 arg2

在flink-bin的bin/flink中，提交一个jar到flink运行

exec "${JAVA_RUN}" $JVM_ARGS $FLINK_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"

这行命令会让我们跳转到CliFrontend类的main方法中

static int mainInternal(final String[] args) {
    EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);

    // 1. find the configuration directory
    final String configurationDirectory = getConfigurationDirectoryFromEnv();

    // 2. load the global configuration
    final Configuration configuration =
            GlobalConfiguration.loadConfiguration(configurationDirectory);

    // 3. load the custom command lines
    final List<CustomCommandLine> customCommandLines =
            loadCustomCommandLines(configuration, configurationDirectory);

    int retCode = INITIAL_RET_CODE;
    try {
        final CliFrontend cli = new CliFrontend(configuration, customCommandLines);
        CommandLine commandLine =
                cli.getCommandLine(
                        new Options(),
                        Arrays.copyOfRange(args, min(args.length, 1), args.length),
                        true);
        Configuration securityConfig = new Configuration(cli.configuration);
        DynamicPropertiesUtil.encodeDynamicProperties(commandLine, securityConfig);
        SecurityUtils.install(new SecurityConfiguration(securityConfig));
        retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));
    } catch (Throwable t) {
        final Throwable strippedThrowable =
                ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
        LOG.error("Fatal error while running command line interface.", strippedThrowable);
        strippedThrowable.printStackTrace();
    }
    return retCode;
}

打印输出一些环境信息

通过flinki conf dir找到conf文件，解析文件

加载flinkyarnSessionCli和DefaultCli

在cli.parseAndRun中会执行run方法，方法里面会解析jar包，拿到主类参数依赖等信息

getPackagedProgram拿到jarFile,userClassPaths,entryPointClassName,configuration,savepointRestoreSettings,args

try (PackagedProgram program = getPackagedProgram(programOptions, effectiveConfiguration)) {
    executeProgram(effectiveConfiguration, program);
}

然后调用execute来执行

protected void executeProgram(final Configuration configuration, final PackagedProgram program)
        throws ProgramInvocationException {
    ClientUtils.executeProgram(
            new DefaultExecutorServiceLoader(), configuration, program, false, false);
}

executeProgram里面通过反射的方式调用运行，mainMethod.invoke(null, (Object) args)

反射是一种在运行时获取和操作类的信息的机制。它允许我们：

获取类的名称、方法、字段等信息。

创建类的实例。

调用类的方法和访问字段。

动态修改类的结构。

想要拿到一个类或者调用某个类的方法，首先需要获取到该类的class对象

通过对象静态属性.class来获取对应的Class对象

Class<TestClass> class = TestClass.class;

Object类中的getClass()方法，适合在有对象示例的情况下使用

Class<TestClass> class = tc.getClass();

只要通过给定类的字符串名称就可以获取该类，更为拓展，使用class类的forName静态方法

Class<TestClass> class = Class.forName("TestClass");

Java反射和new的区别

反射是动态编译，意思就是说只有运行时才会去获得该对象的实例，举例：Spring就是用反射。new是静态编译，在编译的时候所有模块加载入exe

首先new出来的对象我们无法访问其中的私有属性，但是通过反射出来的对象我们可以通过setAccessible()方法来访问其中的私有属性。

在使用new创建一个对象实例的时候必须知道类名，但是通过反射创建对象有时候不需要知道类名也可以
import java.lang.reflect.*;

public class ReflectionExample {
    public static void main(String[] args) throws ClassNotFoundException {
        // 获取类的Class对象
        Class<?> myClass = Class.forName("com.example.MyClass");

        // 获取类的名称
        String className = myClass.getName();
        System.out.println("Class Name: " + className);

        // 获取类的方法
        Method[] methods = myClass.getMethods();
        for (Method method : methods) {
            System.out.println("Method: " + method.getName());
        }

        // 获取类的字段
        Field[] fields = myClass.getDeclaredFields();
        for (Field field : fields) {
            System.out.println("Field: " + field.getName());
        }
    }
}

在ClientFrontUtils中

try {
    program.invokeInteractiveModeForExecution();
} finally {
    ContextEnvironment.unsetAsContext();
    StreamContextEnvironment.unsetAsContext();
}

最终调用自己写的类的main方法

public void invokeInteractiveModeForExecution() throws ProgramInvocationException {
    FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
    try {
        callMainMethod(mainClass, args);
    } finally {
        FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
    }
}

callMainMethod中，得到运行主类的main方法实例， entryClass自己编写的应用程序

try {
    mainMethod = entryClass.getMethod("main", String[].class);
} catch (NoSuchMethodException e) {
    throw new ProgramInvocationException(
            "The class " + entryClass.getName() + " has no main(String[]) method.");
} catch (Throwable t) {
    throw new ProgramInvocationException(
            "Could not look up the main(String[]) method from the class "
                    + entryClass.getName()
                    + ": "
                    + t.getMessage(),
            t);
}