Shulou(Shulou.com)06/01 Report--

This article mainly explains "how to use flink 1.11 to make the sql client support the implementation of sql files", the article explains the content is simple and clear, easy to learn and understand, the following please follow the editor's ideas slowly in depth, together to study and learn "how to use flink 1.11 to make sql clients support the implementation of sql files" bar.

Background

At present, the sql client of flink provides an interactive sql query service, and users can use the sql client to perform some sql batch tasks or flow tasks. But when I want to perform some sql scheduled tasks, flink does not provide a suitable way, so after a comprehensive consideration, I decided to add a'- filename (- f) 'parameter to the sql client, just like the' hive-f abc.sql'', which can perform a batch of sql tasks.

Source code modification

At present, I just want to perform some batch tasks through the sql client, coupled with some design of the flink sql client itself, so the modified sql client supports SET,DDL,INSERT INTO SELECT when executing the sql file. Other statements such as select are not supported for the time being.

The modified method is as follows:

/ home/flink/bin/sql-client.sh embedded-f flink.sql

CliOptionsParser.java

Add an option to the sql client parameter parsing class to parse the-f parameter.

Public static final Option OPTION_FILENAME = Option

.builder ("f")

Required (false)

.longOpt ("filename")

.numberOfArgs (1)

.argName ("the path of the sql file")

.desc ("SQL from files")

.build (); CliOptions.java

Add a variable filename here

Private final String filename

SqlClient.java

Add handling of-filename to SqlClient

If (options.getUpdateStatement ()! = null) {

/ / execute update statement

Final boolean success = cli.submitUpdate (options.getUpdateStatement ())

If (! success) {

Throw new SqlClientException ("Could not submit given SQL update statement to cluster.")

}

} else if (options.getFilename ()! = null) {

Final boolean success = cli.executeFile (options.getFilename ())

If (! success) {

Throw new SqlClientException ("Could not submit given SQL file to cluster.")

}

} else {

Cli.open ()

} SqlClient#executeFile

Add a specific method to execute the sql file, all the sql in the sql file is divided by a semicolon, and then determine what type it is, and call different methods to execute.

Public boolean executeFile (String filename) {

File file = new File (filename)

If (! file.exists ()) {

PrintError ("the file do not exist")

Return false

} else {

String statement = null

Try {

Statement = FileUtils.readFileToString (file)

} catch (IOException e) {

PrintError ("read the sql file error," + e.getMessage ())

Return false

}

String [] sqls = statement.split (";")

For (String sql: sqls) {

If (sql = = null | | ".equals (sql.trim () {

Continue

}

Final Optional parsedStatement = parseCommand (sql)

If (parsedStatement.isPresent ()) {

SqlCommandCall cmdCall = parsedStatement.get ()

Switch (cmdCall.command) {

Case SET:

CallSet (cmdCall)

Break

.

Case INSERT_INTO:

Case INSERT_OVERWRITE:

CallInsert (cmdCall)

Break

Case CREATE_TABLE:

CallDdl (cmdCall.operands [0], CliStrings.MESSAGE_TABLE_CREATED)

Break

.

Throw new SqlClientException ("Unsupported command:" + cmdCall.command)

}

}

}

}

Return true

Thank you for reading, the above is the content of "how to use flink 1.11 to make sql client support the execution of sql files". After the study of this article, I believe you have a deeper understanding of how to use flink 1.11 to make sql client support the execution of sql files, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.