中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

溫馨提示×

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

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

怎么用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

三江| 利津县| 临邑县| 临桂县| 青浦区| 甘孜| 邵阳县| 莆田市| 长阳| 拜泉县| 阿坝县| 阳信县| 三台县| 沁源县| 如东县| 会昌县| 衡阳县| 靖远县| 建宁县| 项城市| 玉溪市| 内江市| 玉林市| 石狮市| 福州市| 饶河县| 元阳县| 桐柏县| 高尔夫| 亚东县| 奉贤区| 文山县| 冀州市| 湛江市| 伊宁县| 舞钢市| 吐鲁番市| 虞城县| 望谟县| 灵璧县| 南宁市|