1.18.2.5.Table API&SQL(查询表、Table API、SQL、混用Table API和SQL、输出表、翻译与执行查询、Blink planner、Old planner)等

冷不防 2022-11-18 01:43 46阅读 0赞

1.18.2.5.查询表  
1.18.2.5.1.Table API  
1.18.2.5.2.SQL  
1.18.2.5.3.混用Table API和SQL  
1.18.2.6.输出表  
1.18.2.7.翻译与执行查询  
1.18.2.7.1.Blink planner  
1.18.2.7.2.Old planner

#### 1.18.2.5.查询表 ####

##### 1.18.2.5.1.Table API #####

Table API 是关于 Scala 和 Java 的集成语言式查询 API。与 SQL 相反，Table API 的查询不是由字符串指定，而是在宿主语言中逐步构建。

Table API 是基于 Table 类的，该类表示一个表（流或批处理），并提供使用关系操作的方法。这些方法返回一个新的 Table 对象，该对象表示对输入 Table 进行关系操作的结果。 一些关系操作由多个方法调用组成，例如 table.groupBy(…).select()，其中 groupBy(…) 指定 table 的分组，而 select(…) 在 table 分组上的投影。  
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;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    public class Demo { 
    
        public static void main(String[] args) { 
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner().inStreamingMode().build();
            // get a TableEnvironment
            TableEnvironment tableEnv = TableEnvironment.create(bsSettings);
    
            // register Orders table
    
            // scan registered Orders table
            Table orders = tableEnv.from("Orders");
            // compute revenue for all customers from France
            Table revenue = orders
                    .filter($("cCountry").isEqual("FRANCE"))
                    .groupBy($("cID"), $("cName")
                            .select($("cID"), $("cName"), $("revenue").sum().as("revSum"));
    
            // emit or convert Table
            // execute query
    
        }
    
    }

##### 1.18.2.5.2.SQL #####

Flink SQL 是基于实现了SQL标准的 Apache Calcite 的。SQL 查询由常规字符串指定。  
文档 SQL 描述了Flink对流处理和批处理表的SQL支持。  
下面的示例演示了如何指定查询并将结果作为 Table 对象返回。

**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;
    
    public class Demo { 
    
        public static void main(String[] args) { 
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner().inStreamingMode().build();
            // get a TableEnvironment
            TableEnvironment tableEnv = TableEnvironment.create(bsSettings);
    
            // register Orders table
    
            // compute revenue for all customers from France
            Table revenue = tableEnv.sqlQuery(
                    "SELECT cID, cName, SUM(revenue) AS revSum " +
                            "FROM Orders " +
                            "WHERE cCountry = 'FRANCE' " +
                            "GROUP BY cID, cName"
            );
    
            // emit or convert Table
            // execute query
    
        }
    
    }

**Scala版本**

package com.toto.learn.sql
    
    import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    object Demo { 
    
      def main(args: Array[String]): Unit = { 
        val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
        val tableEnv: TableEnvironment = TableEnvironment.create(bbSettings)
    
        val revenue = tableEnv.sqlQuery(
          """ |SELECT cID, cName, SUM(revenue) AS revSum |FROM Orders |WHERE cCountry = 'FRANCE' |GROUP BY cID,cName """.stripMargin)
        
        // emit or convert Table
        // execute query
    
      }
    
    }

如下的示例展示了如何指定一个更新查询，将查询的结果插入到已注册的表中。

package com.toto.demo.sql;
    
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.TableEnvironment;
    
    public class Demo { 
    
        public static void main(String[] args) { 
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner().inStreamingMode().build();
            // get a TableEnvironment
            TableEnvironment tableEnv = TableEnvironment.create(bsSettings);
    
            // register "Orders" table
            // register "RevenueFrance" output table
    
            // compute revenue for all customers from France and emit to "RevenueFrance"
            tableEnv.executeSql(
                    "INSERT INTO RevenueFrance " +
                            "SELECT cID, cName, SUM(revenue) AS revSum " +
                            "FROM Orders " +
                            "WHERE cCountry = 'FRANCE' " +
                            "GROUP BY cID, cName"
            );
    
        }
    
    }

##### 1.18.2.5.3.混用Table API和SQL #####

Table API和SQL查询的混用非常简单因为它们都返回 Table 对象：  
可以在SQL查询返回的Table对象上定义Table API查询。  
在TableEnvironment中注册的结果表可以在 SQL查询的FROM子句中引用，通过这种方法就可以在 Table API 查询的结果上定义SQL查询。

#### 1.18.2.6.输出表 ####

Table通过写入 TableSink 输出。TableSink是一个通用接口，用于支持多种文件格式（如 CSV、Apache Parquet、Apache Avro）、存储系统（如 JDBC、Apache HBase、Apache Cassandra、Elasticsearch）或消息队列系统（如Apache Kafka、RabbitMQ）。

批处理Table只能写入 BatchTableSink，而流处理 Table 需要指定写入 AppendStreamTableSink，RetractStreamTableSink 或者 UpsertStreamTableSink。

请参考文档 Table Sources & Sinks （https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/sourceSinks.html）以获取更多关于可用 Sink 的信息以及如何自定义 TableSink。

方法Table.executeInsert(String tableName) 将Table发送至已注册的TableSink。该方法通过名称在catalog中查找TableSink并确认Table schema和TableSink schema一致。

下面的示例演示如何输出 Table：

package com.toto.demo.sql;

import org.apache.flink.table.api.DataTypes;
    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.descriptors.FileSystem;
    import org.apache.flink.table.descriptors.OldCsv;
    import org.apache.flink.table.descriptors.Schema;
    
    public class Demo { 
    
        public static void main(String[] args) { 
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner().inStreamingMode().build();
            // get a TableEnvironment
            TableEnvironment tableEnv = TableEnvironment.create(bsSettings);
    
            // create an output Table
            final Schema schema = new Schema()
                    .field("a", DataTypes.INT())
                    .field("b", DataTypes.STRING())
                    .field("c", DataTypes.BIGINT());
    
            tableEnv.connect(new FileSystem().path("/path/to/file"))
                    .withFormat(new OldCsv().fieldDelimiter("|").deriveSchema())
                    .withSchema(schema)
                    .createTemporaryTable("CsvSinkTable");
    
            // compute a result Table using Table API operators and/or SQL queries
            Table result = ...
            // emit the result Table to the registered TableSink
            result.executeInsert("CsvSinkTable");
            
        }
    
    }

#### 1.18.2.7.翻译与执行查询 ####

两种计划器翻译和执行查询的方式是不同的。

##### 1.18.2.7.1.Blink planner #####

不论输入数据源是流式的还是批式的，Table API 和 SQL 查询都会被转换成 DataStream 程序。查询在内部表示为逻辑查询计划，并被翻译成两个阶段：  
1.优化逻辑执行计划  
2.翻译成DataStream程序  
当 TableEnvironment.executeSql() 被调用时。该方法是用来执行一个 SQL 语句，一旦该方法被调用， SQL 语句立即被翻译。  
当 Table.executeInsert() 被调用时。该方法是用来将一个表的内容插入到目标表中，一旦该方法被调用， TABLE API 程序立即被翻译。  
当 Table.execute() 被调用时。该方法是用来将一个表的内容收集到本地，一旦该方法被调用， TABLE API 程序立即被翻译。  
当 StatementSet.execute() 被调用时。Table （通过 StatementSet.addInsert() 输出给某个 Sink）和 INSERT 语句 （通过调用 StatementSet.addInsertSql()）会先被缓存到 StatementSet 中，StatementSet.execute() 方法被调用时，所有的 sink 会被优化成一张有向无环图。  
当 Table 被转换成 DataStream 时（参阅与 DataStream 和 DataSet API 结合）。转换完成后，它就成为一个普通的 DataStream 程序，并会在调用 StreamExecutionEnvironment.execute() 时被执行。

**注意：从1.11版本开始，sqlUpdate方法和insertInfo方法被废弃，从这两个方法构建的Table程序必须通过StreamTableEnvironment.execute()方法执行，而不能通过StreamExecutionEnvironment.execute()方法来执行。**

##### 1.18.2.7.2.Old planner #####

Table API 和 SQL 查询会被翻译成 DataStream 或者 DataSet 程序， 这取决于它们的输入数据源是流式的还是批式的。查询在内部表示为逻辑查询计划，并被翻译成两个阶段：  
1.优化逻辑执行计划  
2.翻译成DataStream或DataSet程序。

**Table API 或者 SQL 查询在下列情况下会被翻译：**  
当 TableEnvironment.executeSql() 被调用时。该方法是用来执行一个 SQL 语句，一旦该方法被调用， SQL 语句立即被翻译。  
当 Table.executeInsert() 被调用时。该方法是用来将一个表的内容插入到目标表中，一旦该方法被调用， TABLE API 程序立即被翻译。  
当 Table.execute() 被调用时。该方法是用来将一个表的内容收集到本地，一旦该方法被调用， TABLE API 程序立即被翻译。  
当 StatementSet.execute() 被调用时。Table （通过 StatementSet.addInsert() 输出给某个 Sink）和 INSERT 语句 （通过调用 StatementSet.addInsertSql()）会先被缓存到 StatementSet 中，StatementSet.execute() 方法被调用时，所有的 sink 会被优化成一张有向无环图。  
对于 Streaming 而言，当Table 被转换成 DataStream 时（参阅与 DataStream 和 DataSet API 结合）触发翻译。转换完成后，它就成为一个普通的 DataStream 程序，并会在调用 StreamExecutionEnvironment.execute() 时被执行。对于 Batch 而言，Table 被转换成 DataSet 时（参阅与 DataStream 和 DataSet API 结合）触发翻译。转换完成后，它就成为一个普通的 DataSet 程序，并会在调用 ExecutionEnvironment.execute() 时被执行。

**注意** ：从 1.11 版本开始，sqlUpdate 方法 和 insertInto 方法被废弃。对于 Streaming 而言，如果一个 Table 程序是从这两个方法构建出来的，必须通过 StreamTableEnvironment.execute() 方法执行，而不能通过 StreamExecutionEnvironment.execute() 方法执行；对于 Batch 而言，如果一个 Table 程序是从这两个方法构建出来的，必须通过 BatchTableEnvironment.execute() 方法执行，而不能通过 ExecutionEnvironment.execute() 方法执行。