2021年大数据Flink（三十一）：​​​​​​​Table与SQL案例准备 依赖和​​​​​​​程序结构

Posted 2021-05-18 Lansonli

目录

案例准备

依赖

​​​​​​​程序结构

---

​​​​​​​案例准备

依赖

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/

 

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<!-- flink执行计划,这是1.9版本之前的-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>
<!-- blink执行计划,1.11+默认的-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-blink_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

 

● flink-table-common：这个包中主要是包含 Flink Planner 和 Blink Planner一些共用的代码。

● flink-table-api-java：这部分是用户编程使用的 API，包含了大部分的 API。

● flink-table-api-scala：这里只是非常薄的一层，仅和 Table API 的 Expression 和 DSL 相关。

● 两个 Planner：flink-table-planner 和 flink-table-planner-blink。

● 两个 Bridge：flink-table-api-scala-bridge 和 flink-table-api-java-bridge，

Flink Planner 和 Blink Planner 都会依赖于具体的 JavaAPI，也会依赖于具体的 Bridge，通过 Bridge 可以将 API 操作相应的转化为Scala 的 DataStream、DataSet，或者转化为 JAVA 的 DataStream 或者Data Set

 

​​​​​​​程序结构

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/common.html#structure-of-table-api-and-sql-programs

 

 

 

// create a TableEnvironment for specific planner batch or streaming

TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section



// create a Table

tableEnv.connect(...).createTemporaryTable("table1");

// register an output Table

tableEnv.connect(...).createTemporaryTable("outputTable");



// create a Table object from a Table API query

Table tapiResult = tableEnv.from("table1").select(...);



// create a Table object from a SQL query

Table sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");



// emit a Table API result Table to a TableSink, same for SQL result

TableResult tableResult = tapiResult.executeInsert("outputTable");

tableResult...