4.2.1 Flink-流处理框架-Table API 与 SQL-基本概念和简单实例程序

素颜马尾好姑娘i 2022-11-07 11:39 116阅读 0赞

**目录**

1.Table API 与 SQL的基本概念

2.Table API 与 SQL的实例程序

2.1 需要引入pom的依赖

2.2 两种 planner（old & blink）的区别

2.3 实例程序

--------------------

# 1.Table API 与 SQL的基本概念 #

Flink 本身是批流统一的处理框架，所以 Table API 和 SQL，就是批流统一的上层处理 API。 目前功能尚未完善，处于活跃的开发阶段。Table API 是一套内嵌在 Java 和 Scala 语言中的查询 API，它允许我们以非常直观的方式， 组合来自一些关系运算符的查询（比如 select、filter 和 join）。而对于 Flink SQL，就是直接可 以在代码中写 SQL，来实现一些查询（Query）操作。Flink 的 SQL 支持，基于实现了 SQL 标 准的 Apache Calcite（Apache 开源 SQL 解析工具）。无论输入是批输入还是流式输入，在这两套 API 中，指定的查询都具有相同的语义，得到相同的结果。

Table API 与 SQL实际上是一样的效果，看自己更喜欢哪种方式。最近在准备找工作，碰上了一个特别喜欢的部门，主要是做Flink SQL的研究，希望自己好好加油，能够进入到这个团队。加油！！！

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

# 2.Table API 与 SQL的实例程序 #

## 2.1 需要引入pom的依赖 ##

主要是有两个依赖，**一个是planner计划器，另外一个是bridge桥接器**，计划器中包含了桥接器，所以只需要引入这一个依赖即可。阿里巴巴在flink本身的planner计划器之上开发了一个blink框架，我们两个可以同时引入。当我们同时引入了原版本的planner计划器和阿里的blink版本，在1.10版本上默认使用的是原版本planner计划器，在1.11版本之后默认使用的是阿里的blink版本。

flink-table-planner：planner 计划器，是 table API 最主要的部分，提供了运行时环境和生成程序执行计划的 planner；flink-table-api-java-bridge：bridge 桥接器，主要负责 table API 和 DataStream/DataSet API 的连接支持，按照语言分 java 和 scala。这里的两个依赖，是 IDE 环境下运行需要添加的；如果是生产环境，lib 目录下默认已经有了 planner，就只需要有 bridge 就可以了。当然，如果想使用用户自定义函数，或是跟 kafka 做连接，需要有一个 SQL client，这个 包含在 flink-table-common 里。

老版本planner和blink是不兼容的，blink真正做到了批流统一，在blink planner计划器中，无论是批处理程序还是流处理程序，都转化为了dataStream。

<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-planner-blink_2.12</artifactId>
        <version>1.10.1</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_2.12</artifactId>
        <version>1.10.1</version>
    </dependency>

## 2.2 两种 planner（old & blink）的区别 ##

> 1. 批流统一：Blink 将批处理作业，视为流式处理的特殊情况。所以，blink 不支持表和 DataSet 之间的转换，批处理作业将不转换为 DataSet 应用程序，而是跟流处理一样，转换 为 DataStream 程序来处理。
> 
> 2. 因 为 批 流 统 一 ， Blink planner 也 不 支 持 BatchTableSource ， 而 使 用 有 界 的 StreamTableSource 代替。
> 
> 3. Blink planner 只支持全新的目录，不支持已弃用的 ExternalCatalog。
> 
> 4. 旧 planner 和 Blink planner 的 FilterableTableSource 实现不兼容。旧的 planner 会把 PlannerExpressions 下推到 filterableTableSource 中，而 blink planner 则会把 Expressions 下推。
> 
> 5. 基于字符串的键值配置选项仅适用于 Blink planner。
> 
> 6. PlannerConfig 在两个 planner 中的实现不同。
> 
> 7. Blink planner 会将多个 sink 优化在一个 DAG 中（仅在 TableEnvironment 上受支持，而 在 StreamTableEnvironment 上不受支持）。而旧 planner 的优化总是将每一个 sink 放在一个新 的 DAG 中，其中所有 DAG 彼此独立。
> 
> 8. 旧的 planner 不支持目录统计，而 Blink planner 支持。

## 2.3 实例程序 ##

为了符合我们之前学习流的习惯，我们首先创建一个dataStream流，将其转化为表。分别执行tableAPI和SQL方法，查看最后的输出结果。

package com.ucas.apitest.tableapi;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.types.Row;
    
    
    public class TableTest1_Example {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 1. 读取数据
            DataStreamSource<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 2. 转换成POJO
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 3. 创建表环境
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 4. 基于流创建一张表
            Table dataTable = tableEnv.fromDataStream(dataStream);
    
            // 5. 调用table API进行转换操作
            Table resultTable = dataTable.select("id, temperature")
                    .where("id = 'sensor_1'");
    
            // 6. 执行SQL
            tableEnv.createTemporaryView("sensor", dataTable);
            String sql = "select id, temperature from sensor where id = 'sensor_1'";
            Table resultSqlTable = tableEnv.sqlQuery(sql);
    
            tableEnv.toAppendStream(resultTable, Row.class).print("result");
            tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");
    
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/e96d4d60c1b94fa5b340dea4cb2b6155.png