FlinkSQL-- sql-client及源码解析 -- flink-1.13.6_如何代码实现sql-client m -f 这样的命令

一、前言

本文基于flink-1.13.6

SQL Client: Init scripts and Statement Sets

这个版本极大地改进了 SQL 客户端的功能。现在 SQL Client 和 SQL 脚本都支持 通过Java 应用程序执行的几乎所有操作(从 TableEnvironment 以编程方式启动查询)。这意味着 SQL 用户在 SQL 部署中需要的代码少了很多。其中最核心的功能就是支持了 -i 命令用来初始化脚本,-f 命令用来执行 SQL 语句，之前的 YAML 文件这个版本不再支持了,相反更多的是通过 SQL 脚本的方式来配置会话和提交任务.

类似于下面这种方式:

sql-client.sh -i init.sql -f test.sql
1

1.1、 -i 初始化 SQL Client

SET execution.runtime-mode=batch;
SET sql-client.execution.result-mode=TABLEAU;
SET pipeline.name=batch_demo
1
2
3

init.sql 初始化脚本文件支持的功能还非常多,我这里就简单的设置了几个,更多的属性可以参考官网.

使用 -i <init.sql> 选项初始化 SQL Client 会话时，初始化 SQL 文件中允许以下语句：

DDL(CREATE/DROP/ALTER),
USE CATALOG/DATABASE,
LOAD/UNLOAD MODULE,
SET command,
RESET command.
1
2
3
4
5

1.2、-f SQL脚本

create table rate_history ( 
 currency STRING, 
 conversion_rate DECIMAL(32, 2), 
 update_time TIMESTAMP(3),
 WATERMARK FOR update_time AS update_time  
) WITH ( 
 'connector.type' = 'filesystem', 
 'connector.path' = '/tmp/ratesHistory.csv', 
 'format.type' = 'csv' 
);

CREATE TABLE printb
(
  num bigint
)
WITH ('connector' = 'print');

-- 两条sql语句
insert into printb select count(1) from rate_history;
insert into printb select count(1) from rate_history;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

执行：

./bin/sql-client.sh -i test/init.sql -f test/batch.sql 

1
2

查看flink web 页面发现两个job

SQL Client 将每个 INSERT INTO 语句作为单个 Flink 作业执行。但是，由于管道的某些部分可以重复使用，因此有时不是最佳选择。

SQL Client 支持 STATEMENT SET 语法来执行一组 SQL 语句。这是 Table API 中StatementSet 的等效功能。STATEMENT SET 语法包含一个或多个 INSERT INTO 语句。全面优化了STATEMENT SET 块中的所有语句，并将其作为单个 Flink 作业执行。联合优化和执行允许重用常见的中间结果，因此可以显着提高执行多个查询的效率。

STATEMENT SET 的语法格式如下:

BEGIN STATEMENT SET;
  -- one or more INSERT INTO statements
  { INSERT INTO|OVERWRITE <select_statement>; }+
END;

-- 修改上面的sql脚本
-- 两条sql语句
BEGIN STATEMENT SET;
insert into printb select count(1) from rate_history;
insert into printb select count(1) from rate_history;
END;
1
2
3
4
5
6
7
8
9
10
11

.接下来就来看一下底层源码是怎么实现的.

二、源码分析

2.1、从sql-client.sh 找到执行的入口类是 org.apache.flink.table.client.SqlClient

然后来看下 SqlClient 对象属性源码如下:

public class SqlClient { 
    // 标记是否是 embedded 模式
    private final boolean isEmbedded;
    // 提交命令选项
    private final CliOptions options;
    // 用来返回结果的
    private final Supplier<Terminal> terminalFactory;
    // 目前只支持 embedded
    public static final String MODE_EMBEDDED = "embedded";
    public static final String MODE_GATEWAY = "gateway";
	// ...
}
1
2
3
4
5
6
7
8
9
10
11
12

2.2、接着来看 SqlClient 的 main 方法,也就是程序的入口

main 方法里面调用的是 startClient 方法,所以直接来看 startClient 方法的源码:

@VisibleForTesting
protected static void startClient(String[] args, Supplier<Terminal> terminalFactory) {
    final String mode;
    final String[] modeArgs;
    // 设置启动模式默认是 embedded
    if (args.length < 1 || args[0].startsWith("-")) {
        // mode is not specified, use the default `embedded` mode
        mode = MODE_EMBEDDED;
        modeArgs = args;
    } else {
        // mode is specified, extract the mode value and reaming args
        mode = args[0];
        // remove mode
        modeArgs = Arrays.copyOfRange(args, 1, args.length);
    }
    
    switch (mode) {
        case MODE_EMBEDDED:
            // 解析提交命令里的参数
            final CliOptions options = CliOptionsParser.parseEmbeddedModeClient(modeArgs);
            // 打印参数说明
            if (options.isPrintHelp()) {
                CliOptionsParser.printHelpEmbeddedModeClient();
            } else {
                try {
                    // 构建 SqlClient 对象
                    final SqlClient client = new SqlClient(true, options, terminalFactory);
                    client.start();
                } catch (SqlClientException e) {
                    //...
                }
            }
            break;
        case MODE_GATEWAY:
        	// gateway 模式暂时不支持
            throw new SqlClientException("Gateway mode is not supported yet.");
        default:
            CliOptionsParser.printHelpClient();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

2.2.1、解析参数

调用 parseEmbeddedModeClient 方法解析提交命令里面的各种参数.包括我们上面用到的 -i 和 -f 都是在这一步解析并赋值的.

    public static CliOptions parseEmbeddedModeClient(String[] args) {
        try {
            DefaultParser parser = new DefaultParser();
            CommandLine line = parser.parse(EMBEDDED_MODE_CLIENT_OPTIONS, args, true);
            return new CliOptions(
                    line.hasOption(CliOptionsParser.OPTION_HELP.getOpt()),
                    checkSessionId(line),
                    // 解析 -i  初始化文件
                    checkUrl(line, CliOptionsParser.OPTION_INIT_FILE),
                    // 解析 -f sql脚本
                    checkUrl(line, CliOptionsParser.OPTION_FILE),
                    checkUrls(line, CliOptionsParser.OPTION_JAR),
                    checkUrls(line, CliOptionsParser.OPTION_LIBRARY),
                    line.getOptionValue(CliOptionsParser.OPTION_UPDATE.getOpt()),
                    line.getOptionValue(CliOptionsParser.OPTION_HISTORY.getOpt()),
                    getPythonConfiguration(line));
        } catch (ParseException e) {
            throw new SqlClientException(e.getMessage());
        }
    }
    public static final Option OPTION_INIT_FILE =
            Option.builder("i")
                    .required(false)
                    .longOpt("init")
                    .numberOfArgs(1)
                    .argName("initialization file")
                    .desc(
                            "Script file that used to init the session context. "
                                    + "If get error in execution, the sql client will exit. Notice it's not allowed to add query or insert into the init file.")
                    .build();

    public static final Option OPTION_FILE =
            Option.builder("f")
                    .required(false)
                    .longOpt("file")
                    .numberOfArgs(1)
                    .argName("script file")
                    .desc(
                            "Script file that should be executed. In this mode, "
                                    + "the client will not open an interactive terminal.")
                    .build();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

2.2.2、构建 SqlClient

final SqlClient client = new SqlClient(true, options, terminalFactory);
1

2.2.3、启动 SqlClient

client.start();

private void start() {
        if (isEmbedded) {
            // create local executor with default environment
            DefaultContext defaultContext = LocalContextUtils.buildDefaultContext(options);
            // 创建一个 LocalExecutor 对象,用于本地执行程序
            final Executor executor = new LocalExecutor(defaultContext);
            executor.start();

            // Open an new session
            String sessionId = executor.openSession(options.getSessionId());
            try {
                // add shutdown hook
                Runtime.getRuntime()
                        .addShutdownHook(new EmbeddedShutdownThread(sessionId, executor));

                // do the actual work  真正执行 SQL 的地方
                openCli(sessionId, executor);
            } finally {
                executor.closeSession(sessionId);
            }
        }
        else {
            throw new SqlClientException("Gateway mode is not supported yet.");
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2.2.4、真正执行 SQL 的地方是 openCli 方法

  /**
     * Opens the CLI client for executing SQL statements.
     *
     * @param sessionId session identifier for the current client.
     * @param executor executor
     */
    private void openCli(String sessionId, Executor executor) {
        Path historyFilePath;
        if (options.getHistoryFilePath() != null) {
            historyFilePath = Paths.get(options.getHistoryFilePath());
        } else {
            historyFilePath =
                    Paths.get(
                            System.getProperty("user.home"),
                            SystemUtils.IS_OS_WINDOWS ? "flink-sql-history" : ".flink-sql-history");
        }

        boolean hasSqlFile = options.getSqlFile() != null;
        boolean hasUpdateStatement = options.getUpdateStatement() != null;
        if (hasSqlFile && hasUpdateStatement) {
            throw new IllegalArgumentException(//...
        }

        try (CliClient cli = new CliClient(terminalFactory, sessionId, executor, historyFilePath)) {
        	// 执行初始化 SQL -i 参数
            if (options.getInitFile() != null) {
                boolean success = cli.executeInitialization(readFromURL(options.getInitFile()));
                if (!success) { // ...}
            }

            if (!hasSqlFile && !hasUpdateStatement) {
                cli.executeInInteractiveMode();
            } else {
            	// 执行真正的 SQL 文件 -f
                cli.executeInNonInteractiveMode(readExecutionContent());
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

这个里面会先获取 historyFilePath 的路径,然后判断是否存在 -i -f 这两个文件,如果有的话会先调用 executeInitialization 执行初始化的脚本.实际调用的是 executeInitialization#executeFile 方法来执行脚本,executeFile 的源码如下:

private boolean executeFile(String content, ExecutionMode mode) {
    terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_EXECUTE_FILE).toAnsi());
    for (String statement : CliStatementSplitter.splitContent(content)) {
        terminal.writer()
                .println(
                        new AttributedString(String.format("%s%s", prompt, statement))
                                .toString());
        terminal.flush();
        // 执行 
        if (!executeStatement(statement, mode)) {
            // cancel execution when meet error or ctrl + C;
            return false;
        }
    }
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

其实不管是 -i 还是 -f 最终都会调用 executeFile 这个方法去解析脚本里的内容并且执行,这里方法里面先调用 splitContent 方法去做解析.

public static List<String> splitContent(String content) {
    List<String> statements = new ArrayList<>();
    List<String> buffer = new ArrayList<>();
    for (String line : content.split("\n")) {
        if (isEndOfStatement(line)) {
            buffer.add(line);
            statements.add(String.join("\n", buffer));
            buffer.clear();
        } else {
            buffer.add(line);
        }
    }
    if (!buffer.isEmpty()) {
        statements.add(String.join("\n", buffer));
    }
    return statements;
}
private static boolean isEndOfStatement(String line) {
    return line.replaceAll(MASK, "").trim().endsWith(";");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

其实就是一行一行的读取初始化脚本和 SQL 脚本里面的内容,然后放到一个 List 里面.然后循环这个 List 调用 executeStatement 方法去执行 SQL 脚本.

// 执行 SQL 脚本.
private boolean executeStatement(String statement, ExecutionMode executionMode) {
    try {
        final Optional<Operation> operation = parseCommand(statement);
        operation.ifPresent(op -> callOperation(op, executionMode));
    } catch (SqlExecutionException e) {
        printExecutionException(e);
        return false;
    }
    return true;
}
1
2
3
4
5
6
7
8
9
10
11

执行之前会先对 SQL 做一个清洗,具体逻辑在 parseCommand 方法中.

// 其实就是把 SQL 后面的 ; 去掉,并在遇到 bad case 的时候返回空.然后调用 parseStatement 方法将 SQL 语句解析成 Operation,后面的过程就跟 Flink SQL 翻译成代码的过程差不多.就不在往后面跟了.
private Optional<Operation> parseCommand(String stmt) {
    // normalize
    stmt = stmt.trim();
    // remove ';' at the end
    if (stmt.endsWith(";")) {
        stmt = stmt.substring(0, stmt.length() - 1).trim();
    }
    // meet bad case, e.g ";\n"
    if (stmt.trim().isEmpty()) {
        return Optional.empty();
    }
    
    Operation operation = executor.parseStatement(sessionId, stmt);
    return Optional.of(operation);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

-f 参数调用的是 executeInNonInteractiveMode 方法,实际也会调用 executeFile 方法,跟 -i 的执行逻辑是一样的.这里就不再分析了.

另外当前的 SQL Client 仅支持嵌入式模式(也就是 embedded 模式)。将来，社区计划通过提供基于 REST 的SQL 客户端网关来扩展其功能，有关更多信息，请参见 FLIP-24 和 FLIP-91。