91超碰碰碰碰久久久久久综合_超碰av人澡人澡人澡人澡人掠_国产黄大片在线观看画质优化_txt小说免费全本

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

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

發布時間:2021-12-31 10:38:30 來源:億速云 閱讀:329 作者:iii 欄目:大數據

這篇文章主要講解了“怎么用flink 1.11使sql客戶端支持執行sql文件”,文中的講解內容簡單清晰,易于學習與理解,下面請大家跟著小編的思路慢慢深入,一起來研究和學習“怎么用flink 1.11使sql客戶端支持執行sql文件”吧!

背景

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

源碼修改

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

修改后執行的方式為:

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

CliOptionsParser.java

在這個sql 客戶端參數解析類里添加一個選項,用于解析-f參數。

 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

添加具體的執行sql文件的方法,sql文件里的所有sql以分號切分,然后分別判斷是什么類型,調用不同的方法來執行。

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

感謝各位的閱讀,以上就是“怎么用flink 1.11使sql客戶端支持執行sql文件”的內容了,經過本文的學習后,相信大家對怎么用flink 1.11使sql客戶端支持執行sql文件這一問題有了更深刻的體會,具體使用情況還需要大家實踐驗證。這里是億速云,小編將為大家推送更多相關知識點的文章,歡迎關注!

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

娱乐| 措勤县| 五台县| 嘉鱼县| 丘北县| 兴仁县| 丰城市| 壶关县| 玉田县| 巫溪县| 龙井市| 化德县| 杂多县| 黎川县| 织金县| 肃北| 肃宁县| 遂宁市| 苏尼特右旗| 雷山县| 隆尧县| 阳西县| 聂拉木县| 敦煌市| 柯坪县| 满洲里市| 句容市| 乾安县| 房产| 曲沃县| 武邑县| 潼关县| 莱西市| 峨边| 株洲市| 宁安市| 奉节县| 肇州县| 南部县| 北宁市| 吉水县|