1.18.2.Table API&SQL(概念与通用API、两种计划器（Planner）的主要区别、创建 TableEnvironment、临时表、永久表、创建表、虚拟表、Connector 等)

谁践踏了优雅 2022-11-18 01:43 59阅读 0赞

1.18.2.概念与通用API  
1.18.2.1.两种计划器（Planner）的主要区别：  
1.18.2.2.Table API和SQL程序的结构  
1.18.2.3.创建 TableEnvironment  
1.18.2.4.在Catalog中创建表  
1.18.2.4.1.临时表（Temporary Table）和永久表（Permanent Table）  
1.18.2.4.1.1.屏蔽（Shadowing）  
1.18.2.4.2.创建表  
1.18.2.4.2.1.虚拟表  
1.18.2.4.2.2.Connector Tables  
1.18.2.4.3.扩展表标识符

### 1.18.2.概念与通用API ###

Table API 和 SQL 集成在同一套 API 中。这套 API 的核心概念是Table，用作查询的输入和输出。本文介绍了 Table API 和 SQL 查询程序的通用结构、如何注册 Table 、如何查询 Table 以及如何输出 Table。

#### 1.18.2.1.两种计划器（Planner）的主要区别： ####

1.Blink将批处理作业视作流处理的一种特例。严格来说，Table 和 DataSet 之间不支持相互转换，并且批处理作业也不会转换成 DataSet 程序而是转换成 DataStream 程序，流处理作业也一样。  
2.Blink 计划器不支持 BatchTableSource，而是使用有界的 StreamTableSource 来替代。  
3.旧计划器和 Blink 计划器中 FilterableTableSource 的实现是不兼容的。旧计划器会将 PlannerExpression 下推至 FilterableTableSource，而 Blink 计划器则是将 Expression 下推。  
4.基于字符串的键值配置选项仅在 Blink 计划器中使用。（详情参见 配置 ）  
5.PlannerConfig 在两种计划器中的实现（CalciteConfig）是不同的。  
6.Blink 计划器会将多sink（multiple-sinks）优化成一张有向无环图（DAG），TableEnvironment 和 StreamTableEnvironment 都支持该特性。旧计划器总是将每个sink都优化成一个新的有向无环图，且所有图相互独立。  
7.旧计划器目前不支持 catalog 统计数据，而 Blink 支持。

#### 1.18.2.2.Table API和SQL程序的结构 ####

所有用于批处理和流处理的 Table API 和 SQL 程序都遵循相同的模式。下面的代码示例展示了 Table API 和 SQL 程序的通用结构。

其中pom.xml添加的依赖如下：

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_2.11</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner_2.11</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.11</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.11</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-common</artifactId>
        <version>1.12.0</version>
        
    </dependency>
    
    <build>
        <plugins>
            
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.0</version>
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                    <encoding>UTF-8</encoding>
                    <compilerVersion>${maven.compiler.source}</compilerVersion>
                    <showDeprecation>true</showDeprecation>
                    <showWarnings>true</showWarnings>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.12.4</version>
                <configuration>
                    <skipTests>${maven.test.skip}</skipTests>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.rat</groupId>
                <artifactId>apache-rat-plugin</artifactId>
                <version>0.12</version>
                <configuration>
                    <excludes>
                        <exclude>README.md</exclude>
                    </excludes>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-checkstyle-plugin</artifactId>
                <version>2.17</version>
                <executions>
                    <execution>
                        <id>verify</id>
                        <phase>verify</phase>
                        <configuration>
                            <configLocation>style/rmq_checkstyle.xml</configLocation>
                            <encoding>UTF-8</encoding>
                            <consoleOutput>true</consoleOutput>
                            <failsOnError>true</failsOnError>
                            <includeTestSourceDirectory>false</includeTestSourceDirectory>
                            <includeTestResources>false</includeTestResources>
                        </configuration>
                        <goals>
                            <goal>check</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-javadoc-plugin</artifactId>
                <version>2.10.4</version>
                <configuration>
                    <aggregate>true</aggregate>
                    <reportOutputDirectory>javadocs</reportOutputDirectory>
                    <locale>en</locale>
                </configuration>
            </plugin>
            
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.1.6</version>
                <configuration>
                    <scalaCompatVersion>2.11</scalaCompatVersion>
                    <scalaVersion>2.11.12</scalaVersion>
                    <encoding>UTF-8</encoding>
                </configuration>
                <executions>
                    <execution>
                        <id>compile-scala</id>
                        <phase>compile</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>test-compile-scala</id>
                        <phase>test-compile</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.6</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            
                            <mainClass></mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

**Java代码如下：**

package com.toto.demo.sql;
    
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    import org.apache.flink.table.api.TableResult;
    
    /** * @author tuzuoquan * * @description * * TableAPIAndSQLStructure * * @date 2021/1/28 10:27 **/
    public class TableAPIAndSQLStructure { 
    
        public static void main(String[] args) { 
    
            //Flink STREAMING QUERY
            EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
            StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // create a TableEnvironment for specific planner batch or streaming
            //StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
            //或者使用下面的方式
            TableEnvironment tableEnv = TableEnvironment.create(fsSettings);
    
            // create an input Table
            tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");
            // register an output Table
            tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )");
    
            // create a Table object from a Table API query
            Table table2 = tableEnv.from("table1").select(...);
            // create a Table object from a SQL query
            Table table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");
    
            // emit a Table API result Table to a TableSink, same for SQL result
            TableResult tableResult = table2.executeInsert("outputTable");
            tableResult...
    
        }
    
    }

**scala代码如下：**

package com.toto.learn.sql
    
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    object TableAPIAndSQLStructure { 
    
      def main(args: Array[String]): Unit = { 
    
        //FLINK STREAMING QUERY
        val fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
        val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment
        //val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)
        // 或使用下面的创建方式
        val tableEnv = TableEnvironment.create(fsSettings)
    
        // create a TableEnvironment for specific planner batch or streaming
        //val tableEnv = ... // see "Create a TableEnvironment" section
    
        // create an input Table
        tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )")
        // register an output Table
        tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )")
    
        // create a Table from a Table API query
        val table2 = tableEnv.from("table1").select(...)
        // create a Table from a SQL query
        val table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
    
        // emit a Table API result Table to a TableSink, same for SQL result
        val tableResult = table2.executeInsert("outputTable")
        tableResult...
    
      }
    
    }

#### 1.18.2.3.创建 TableEnvironment ####

TableEnvironment是Table API和SQL的核心概念。它负责：  
1.在内部的catalog中注册Table  
2.注册外部的catalog  
3.加载可插拔模块  
4.执行SQL查询  
5.注册自定义函数 （scalar、table 或 aggregation）  
6.将 DataStream 或 DataSet 转换成 Table  
7.持有对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用。

Table 总是与特定的 TableEnvironment 绑定。不能在同一条查询中使用不同 TableEnvironment 中的表，例如，对它们进行 join 或 union 操作。

TableEnvironment 可以通过静态方法 BatchTableEnvironment.create() 或者 StreamTableEnvironment.create() 在 StreamExecutionEnvironment 或者 ExecutionEnvironment 中创建，TableConfig 是可选项。TableConfig可用于配置TableEnvironment或定制的查询优化和转换过程(参见 查询优化（https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/common.html\#query-optimization）)。

请确保选择与你的编程语言匹配的特定的计划器BatchTableEnvironment/StreamTableEnvironment。  
如果两种计划器的 jar 包都在 classpath 中（默认行为），你应该明确地设置要在当前程序中使用的计划器。

**JAVA版本**

// **********************
    // FLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
    StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
    // or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);
    
    // ******************
    // FLINK BATCH QUERY
    // ******************
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
    
    ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
    BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);
    
    // **********************
    // BLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
    StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
    // or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
    
    // ******************
    // BLINK BATCH QUERY
    // ******************
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.TableEnvironment;
    
    EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
    TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);

**SCALA版本**

// **********************
    // FLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    val fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)
    // or val fsTableEnv = TableEnvironment.create(fsSettings)
    
    // ******************
    // FLINK BATCH QUERY
    // ******************
    import org.apache.flink.api.scala.ExecutionEnvironment
    import org.apache.flink.table.api.bridge.scala.BatchTableEnvironment
    
    val fbEnv = ExecutionEnvironment.getExecutionEnvironment
    val fbTableEnv = BatchTableEnvironment.create(fbEnv)
    
    // **********************
    // BLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
    // or val bsTableEnv = TableEnvironment.create(bsSettings)
    
    // ******************
    // BLINK BATCH QUERY
    // ******************
    import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
    val bbTableEnv = TableEnvironment.create(bbSettings)

注意：如果/lib目录中只有一种计划器的 jar 包，则可以使用useAnyPlanner（python 使用 use any\_u\_planner）创建 EnvironmentSettings。

#### 1.18.2.4.在Catalog中创建表 ####

TableEnvironment 维护着一个由标识符（identifier）创建的表 catalog 的映射。标识符由三个部分组成：catalog 名称、数据库名称以及对象名称。如果 catalog 或者数据库没有指明，就会使用当前默认值。（参见表标识符扩展章节中的例子）。

Table 可以是虚拟的（视图 VIEWS）也可以是常规的（表 TABLES）。视图VIEWS可以从已经存在的Table中创建，一般是Table API或者SQL的查询结果。表TABLES描述的是外部数据，例如文件、数据库表或者消息队列。

##### 1.18.2.4.1.临时表（Temporary Table）和永久表（Permanent Table） #####

表可以是临时的，并与单个 Flink 会话（session）的生命周期相关，也可以是永久的，并且在多个 Flink 会话和群集（cluster）中可见。

永久表需要 catalog（例如 Hive Metastore）以维护表的元数据。一旦永久表被创建，它将对任何连接到 catalog 的 Flink 会话可见且持续存在，直至被明确删除。

###### 1.18.2.4.1.1.屏蔽（Shadowing） ######

可以使用与已存在的永久表相同的标识符去注册临时表。临时表会屏蔽永久表，并且只要临时表存在，永久表就无法访问。所有使用该标识符的查询都将作用于临时表。

这可能对实验（experimentation）有用。它允许先对一个临时表进行完全相同的查询，例如只有一个子集的数据，或者数据是不确定的。一旦验证了查询的正确性，就可以对实际的生产表进行查询。

##### 1.18.2.4.2.创建表 #####

###### 1.18.2.4.2.1.虚拟表 ######

在 SQL 的术语中，Table API 的对象对应于视图（虚拟表）。它封装了一个逻辑查询计划。它可以通过以下方法在 catalog 中创建：

**Scala代码**

package com.toto.learn.sql
    
    import org.apache.flink.table.api.{ EnvironmentSettings, Table, TableEnvironment}
    
    object CreateTable { 
    
      def main(args: Array[String]): Unit = { 
        val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
        val tableEnv = TableEnvironment.create(bbSettings)
    
        // table is the result of a simple projection query 
        val projTable: Table = tableEnv.from("X").select(...)
        tableEnv.createTemporaryView("projectedTable",projTable)
    
      }
      
    }

注意：从传统数据库系统的角度来看，Table对象与VIEW视图非常像。也就是，定义了Table的查询是没有被优化的，而且会被内嵌到另一个引用了这个注册了的Table的查询中。如果多个查询都引用了同一个注册了的Table，那么它会被内嵌每个查询中被执行多次，也就是说注册了的Table的结果不会被共享（注：Blink计划器的TableEnvironment会被优化成只执行了一次）。

###### 1.18.2.4.2.2.Connector Tables ######

另外一个方式去创建 TABLE 是通过 connector 声明。Connector 描述了存储表数据的外部系统。存储系统例如 Apache Kafka 或者常规的文件系统都可以通过这种方式来声明。

**java代码：**

tableEnvironment
      .connect(...)
      .withFormat(...)
      .withSchema(...)
      .inAppendMode()
      .createTemporaryTable("MyTable")

**Scala代码：**

tableEnvironment
      .connect(...)
      .withFormat(...)
      .withSchema(...)
      .inAppendMode()
      .createTemporaryTable("MyTable")

##### 1.18.2.4.3.扩展表标识符 #####

表总是通过三元标识符注册，包括 catalog 名、数据库名和表名。

用户可以指定一个 catalog 和数据库作为 “当前catalog” 和”当前数据库”。有了这些，那么刚刚提到的三元标识符的前两个部分就可以被省略了。如果前两部分的标识符没有指定， 那么会使用当前的 catalog 和当前数据库。用户也可以通过 Table API 或 SQL 切换当前的 catalog 和当前的数据库。

标识符遵循 SQL 标准，因此使用时需要用反引号（\`）进行转义。  
**Java代码**

package com.toto.demo.sql;
    
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    
    /** * @author tuzuoquan * @version 1.0 * @ClassName * @description TODO * @date 2021/1/29 10:24 **/
    public class Demo { 
    
        public static void main(String[] args) { 
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner().inStreamingMode().build();
            // get a TableEnvironment
            TableEnvironment tableEnv = TableEnvironment.create(bsSettings);
    
            tableEnv.useCatalog("custom_catalog");
            tableEnv.useDatabase("custom_database");
    
            Table table = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");
    
            // register the view named 'exampleView' in the catalog named 'custom_catalog'
            // in the database named 'custom_database'
            tableEnv.createTemporaryView("exampleView", table);
    
            // register the view named 'exampleView' in the catalog named 'custom_catalog'
            // in the database named 'other_database'
            tableEnv.createTemporaryView("other_database.exampleView", table);
    
            // register the view named 'example.View' in the catalog named 'custom_catalog'
            // in the database named 'custom_database'
            tableEnv.createTemporaryView("`example.View`", table);
    
            // register the view named 'exampleView' in the catalog named 'other_catalog'
            // in the database named 'other_database'
            tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table);
    
        }
    
    }

**Scala代码**

package com.toto.learn.sql
    
    import org.apache.flink.table.api.{ EnvironmentSettings, Table, TableEnvironment}
    
    object Demo { 
    
      def main(args: Array[String]): Unit = { 
        val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
        val tableEnv: TableEnvironment = TableEnvironment.create(bbSettings)
    
        tableEnv.useCatalog("custom_catalog")
        tableEnv.useDatabase("custom_database")
    
        val table: Table = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
    
        // register the view named 'exampleView' in the catalog named 'custom_catalog'
        // in the database named 'custom_database'
        tableEnv.createTemporaryView("exampleView", table)
    
        // register the view named 'exampleView' in the catalog named 'custom_catalog'
        // in the database named 'other_database'
        tableEnv.createTemporaryView("other_database.exampleView", table)
    
        // register the view named 'example.View' in the catalog named 'custom_catalog'
        // in the database named 'custom_database'
        tableEnv.createTemporaryView("`example.View`", table)
    
        // register the view named 'exampleView' in the catalog named 'other_catalog'
        // in the database named 'other_database'
        tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table)
    
      }
    
    }