十分鐘入門Fink SQL

作者：大數據老哥 2020-12-17 06:48:21

本篇文章主要講解了Flink SQL 入門操作，后面我會分享一些關于Flink SQL連接Kafka、輸出到kafka、MySQL等

前言

Flink 本身是批流統一的處理框架，所以 Table API 和 SQL，就是批流統一的上層處理 API。目前功能尚未完善，處于活躍的開發階段。 Table API 是一套內嵌在 Java 和 Scala 語言中的查詢 API，它允許我們以非常直觀的方式，組合來自一些關系運算符的查詢(比如 select、filter 和 join)。而對于 Flink SQL，就是直接可以在代碼中寫 SQL，來實現一些查詢(Query)操作。Flink 的 SQL 支持，基于實現了 SQL 標準的 Apache Calcite(Apache 開源 SQL 解析工具)。圖片

1、導入所需要的的依賴包

<dependency> 
          <groupId>org.apache.flink</groupId> 
          <artifactId>flink-table-planner_2.12</artifactId> 
          <version>1.10.1</version> 
      </dependency> 
      <dependency> 
          <groupId>org.apache.flink</groupId> 
          <artifactId>flink-table-api-scala-bridge_2.12</artifactId> 
          <version>1.10.1</version> 
      </dependency> 
      <dependency> 
          <groupId>org.apache.flink</groupId> 
          <artifactId>flink-csv</artifactId> 
          <version>1.10.1</version> 
     </dependency>

flink-table-planner：planner 計劃器，是 table API 最主要的部分，提供了運行時環境和生成程序執行計劃的 planner; flink-table-api-scala-bridge：bridge 橋接器，主要負責 table API 和 DataStream/DataSet API的連接支持，按照語言分 java 和 scala。

這里的兩個依賴，是 IDE 環境下運行需要添加的;如果是生產環境，lib 目錄下默認已經有了 planner，就只需要有 bridge 就可以了。

當然，如果想使用用戶自定義函數，或是跟 kafka 做連接，需要有一個 SQL client，這個包含在 flink-table-common 里。

2、兩種 planner(old& blink)的區別

批流統一：Blink 將批處理作業，視為流式處理的特殊情況。所以，blink 不支持表和DataSet 之間的轉換，批處理作業將不轉換為 DataSet 應用程序，而是跟流處理一樣，轉換為 DataStream 程序來處理。
因為批流統一， Blink planner 也不支持 BatchTableSource ，而使用有界的
Blink planner 只支持全新的目錄，不支持已棄用的 ExternalCatalog。
舊 planner 和 Blink planner 的 FilterableTableSource 實現不兼容。舊的 planner 會把PlannerExpressions 下推到 filterableTableSource 中，而 blink planner 則會把 Expressions 下推。
基于字符串的鍵值配置選項僅適用于 Blink planner。
PlannerConfig 在兩個 planner 中的實現不同。
Blink planner 會將多個 sink 優化在一個 DAG 中(僅在 TableEnvironment 上受支持，而在 StreamTableEnvironment 上不受支持)。而舊 planner 的優化總是將每一個 sink 放在一個新的 DAG 中，其中所有 DAG 彼此獨立。
舊的 planner 不支持目錄統計，而 Blink planner 支持。

3、表(Table)的概念

TableEnvironment 可以注冊目錄 Catalog，并可以基于 Catalog 注冊表。它會維護一個Catalog-Table 表之間的 map。表(Table)是由一個標識符來指定的，由 3 部分組成：Catalog 名、數據庫(database)名和對象名(表名)。如果沒有指定目錄或數據庫，就使用當前的默認值。

4、連接到文件系統(Csv 格式)

連接外部系統在 Catalog 中注冊表，直接調用 tableEnv.connect()就可以，里面參數要傳入一個 ConnectorDescriptor，也就是 connector 描述器。對于文件系統的 connector 而言，flink內部已經提供了，就叫做 FileSystem()。

5、測試案例 (新)

需求：將一個txt文本文件作為輸入流讀取數據過濾id不等于sensor_1的數據實現思路：首先我們先構建一個table的env環境通過connect提供的方法來讀取數據然后設置表結構將數據注冊為一張表就可進行我們的數據過濾了(使用sql或者流處理方式進行解析)

準備數據

sensor_1,1547718199,35.8 
sensor_6,1547718201,15.4 
sensor_7,1547718202,6.7 
sensor_10,1547718205,38.1 
sensor_1,1547718206,32 
sensor_1,1547718208,36.2 
sensor_1,1547718210,29.7 
sensor_1,1547718213,30.9

代碼實現

import org.apache.flink.streaming.api.scala._ 
import org.apache.flink.table.api.{DataTypes} 
import org.apache.flink.table.api.scala._ 
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema} 
 
/** 
 * @Package 
 * @author 大數據老哥 
 * @date 2020/12/12 21:22 
 * @version V1.0 
 *          第一個Flinksql測試案例 
 */ 
 
object FlinkSqlTable { 
  def main(args: Array[String]): Unit = { 
    // 構建運行流處理的運行環境 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    // 構建table環境 
    val tableEnv = StreamTableEnvironment.create(env) 
     //通過 connect 讀取數據 
    tableEnv.connect(new FileSystem().path("D:\\d12\\Flink\\FlinkSql\\src\\main\\resources\\sensor.txt")) 
      .withFormat(new Csv()) //設置類型 
      .withSchema(new Schema() // 給數據添加元數信息 
        .field("id", DataTypes.STRING()) 
        .field("time", DataTypes.BIGINT()) 
        .field("temperature", DataTypes.DOUBLE()) 
      ).createTemporaryTable("inputTable")  // 創建一個臨時表 
     
    val resTable = tableEnv.from("inputTable") 
      .select("*").filter('id === "sensor_1") 
    // 使用sql的方式查詢數據 
    var resSql = tableEnv.sqlQuery("select * from inputTable where id='sensor_1'") 
    // 將數據轉為流進行輸出 
    resTable.toAppendStream[(String, Long, Double)].print("resTable") 
    resSql.toAppendStream[(String, Long, Double)].print("resSql") 
 
    env.execute("FlinkSqlWrodCount") 
  } 
}

6、TableEnvironment 的作用

注冊 catalog
在內部 catalog 中注冊表
執行 SQL 查詢
注冊用戶自定義函數
注冊用戶自定義函數
保存對 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

在創建 TableEnv 的時候，可以多傳入一個 EnvironmentSettings 或者 TableConfig 參數，可以用來配置 TableEnvironment 的一些特性。

7、老版本創建流處理批處理

7.1老版本流處理

val settings = EnvironmentSettings.newInstance() 
.useOldPlanner() // 使用老版本 planner 
.inStreamingMode() // 流處理模式 
.build() 
val tableEnv = StreamTableEnvironment.create(env, settings)

7.2 老版本批處理

val batchEnv = ExecutionEnvironment.getExecutionEnvironment  
val batchTableEnv = BatchTableEnvironment.create(batchEnv)

7.3 blink 版本的流處理環境

val bsSettings = EnvironmentSettings.newInstance() 
.useBlinkPlanner() 
.inStreamingMode().build() 
val bsTableEnv = StreamTableEnvironment.create(env, bsSettings)

7.4 blink 版本的批處理環境

val bbSettings = EnvironmentSettings.newInstance() 
.useBlinkPlanner() 
.inBatchMode().build() 
val bbTableEnv = TableEnvironment.create(bbSettings)

總結：

本篇文章主要講解了Flink SQL 入門操作，后面我會分享一些關于Flink SQL連接Kafka、輸出到kafka、MySQL等

本文轉載自微信公眾號「大數據老哥」，可以通過以下二維碼關注。轉載本文請聯系大數據老哥公眾號。

責任編輯：武曉燕來源：大數據老哥