怎么用flink 1.11使sql客戶端支持執(zhí)行sql文件

背景

目前flink的sql客戶端提供了一種交互式的sql查詢服務(wù)，用戶可以使用sql客戶端執(zhí)行一些sql的批任務(wù)或者流任務(wù)。但是當(dāng)我想執(zhí)行一些sql的定時任務(wù)時，flink卻沒有提供一個合適的方式，所以綜合考慮了一下，我決定在sql的客戶端基礎(chǔ)上給加一個 '-filename (-f)' 參數(shù)，就像類似'hive -f abc.sql' 一樣，可以執(zhí)行一批sql任務(wù)。 

源碼修改

目前我只是想通過sql客戶端執(zhí)行一些批任務(wù)，再加上flink sql 客戶端本身的一些設(shè)計，所以目前修改后的sql client 執(zhí)行sql文件的時候支持 SET,DDL，INSERT INTO SELECT ...等語句，其他比如select暫不支持。

修改后執(zhí)行的方式為：

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

CliOptionsParser.java

在這個sql 客戶端參數(shù)解析類里添加一個選項，用于解析-f參數(shù)。

 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

在這里添加一個變量filename

private final String filename;
  
  

SqlClient.java

在SqlClient里添加對于-filename的處理

  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

添加具體的執(zhí)行sql文件的方法，sql文件里的所有sql以分號切分，然后分別判斷是什么類型，調(diào)用不同的方法來執(zhí)行。

 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<SqlCommandCall> 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;
 }