4.2.2 Flink-流处理框架-Table API 与 SQL-基本程序结构（创建表环境+在 Catalog 中注册表）

Love The Way You Lie 2022-11-07 11:42 171阅读 0赞

**目录**

1.基本程序结构

2.创建表环境

2.1 配置老版本 planner

2.2 配置 blink planner

3.在 Catalog 中注册表

3.1 表的概念

3.2 连接到文件系统（Csv 格式）

3.3 连接到 Kafka

4.代码示例

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

# 1.基本程序结构 #

Table API 和 SQL 的程序结构，与流式处理的程序结构类似；也可以近似地认为有这么 几步：首先创建执行环境，然后定义 source、transform 和 sink。具体操作流程如下：

StreamTableEnvironment tableEnv = ... // 创建表的执行环境
    
    // 创建一张表，用于读取数据
    tableEnv.connect(...).createTemporaryTable("inputTable");
    
    // 注册一张表，用于把计算结果输出
    tableEnv.connect(...).createTemporaryTable("outputTable");
    
    // 通过 Table API 查询算子，得到一张结果表
    Table result = tableEnv.from("inputTable").select(...);
    
    // 通过 SQL 查询语句，得到一张结果表
    Table sqlResult = tableEnv.sqlQuery("SELECT ... FROM inputTable ...");
    
    // 将结果表写入输出表中
    result.insertInto("outputTable");

# 2.创建表环境 #

创建表环境最简单的方式，就是基于流处理执行环境，调 create 方法直接创建：

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

表环境（TableEnvironment）是 flink 中集成 Table API & SQL 的核心概念。它负责:⚫ **注册 catalog** ⚫ **在内部 catalog 中注册表** ⚫ **执行 SQL 查询** ⚫ **注册用户自定义函数UDF** ⚫ 将 DataStream 或 DataSet 转换为表 ⚫ 保存对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

在创建 TableEnv 的时候，可以多传入一个 EnvironmentSettings 或者 TableConfig 参数， 可以用来配置 TableEnvironment 的一些特性。

## 2.1 配置老版本 planner ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 2.2 配置 blink planner ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

# 3.在 Catalog 中注册表 #

## 3.1 表的概念 ##

TableEnvironment 可以注册目录 Catalog，并可以基于 Catalog 注册表。它会维护一个 Catalog-Table 表之间的 map。

表（Table）是由一个“标识符”来指定的，由 3 部分组成：Catalog 名、数据库（database） 名和对象名（表名）。如果没有指定目录或数据库，就使用当前的默认值。

表可以是常规的（Table，表），或者虚拟的（View，视图）。常规表（Table）一般可以用来描述外部数据，比如文件、数据库表或消息队列的数据，也可以直接从 DataStream 转 换而来。视图可以从现有的表中创建，通常是 table API 或者 SQL 查询的一个结果。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

## 3.2 连接到文件系统（Csv 格式） ##

连接外部系统在 Catalog 中注册表，直接调用 tableEnv.connect()就可以，里面参数要传入一个 ConnectorDescriptor，也就是 connector 描述器。对于文件系统的 connector 而言，flink 内部已经提供了，就叫做 FileSystem()。代码如下：

tableEnv
        .connect( new FileSystem().path("sensor.txt")) // 定义表数据来源，外部连接
        .withFormat(new OldCsv()) // 定义从外部系统读取数据之后的格式化方法 
        .withSchema( new Schema()
            .field("id", DataTypes.STRING())
            .field("timestamp", DataTypes.BIGINT())
            .field("temperature", DataTypes.DOUBLE())
         ) // 定义表结构
        .createTemporaryTable("inputTable"); // 创建临时表

这是旧版本的 csv 格式描述器。由于它是非标的，跟外部系统对接并不通用，所以将被 弃用，以后会被一个符合 RFC-4180 标准的新 format 描述器取代。新的描述器就叫 Csv()，但 flink 没有直接提供，需要引入依赖 flink-csv：

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-csv</artifactId>
        <version>1.10.1</version>
    </dependency>

代码非常类似，只需要把 withFormat 里的 OldCsv 改成 Csv 就可以了。

## 3.3 连接到 Kafka ##

kafka 的连接器 flink-kafka-connector 中，1.10 版本的已经提供了 Table API 的支持。我们 可以在 connect 方法中直接传入一个叫做 Kafka 的类，这就是 kafka 连接器的描述器 ConnectorDescriptor。

tableEnv.connect(
        new Kafka()
            .version("0.11") // 定义 kafka 的版本
            .topic("sensor") // 定义主题 .property("zookeeper.connect", "localhost:2181")
            .property("bootstrap.servers", "localhost:9092")
        )
        .withFormat(new Csv())
        .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
    )
    .createTemporaryTable("kafkaInputTable");

当然也可以连接到 ElasticSearch、MySql、HBase、Hive 等外部系统，实现方式基本上是类似的。

# 4.代码示例 #

package com.atguigu.apitest.tableapi;
    
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    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.api.java.BatchTableEnvironment;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.descriptors.Csv;
    import org.apache.flink.table.descriptors.FileSystem;
    import org.apache.flink.table.descriptors.OldCsv;
    import org.apache.flink.table.descriptors.Schema;
    import org.apache.flink.types.Row;
    
    public class TableTest2_CommonApi {
        public static void main(String[] args) throws Exception{
            // 1. 创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 1.1 基于老版本planner的流处理
            EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance()
                    .useOldPlanner()
                    .inStreamingMode()
                    .build();
            StreamTableEnvironment oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings);
    
            // 1.2 基于老版本planner的批处理
            ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment oldBatchTableEnv = BatchTableEnvironment.create(batchEnv);
    
            // 1.3 基于Blink的流处理
            EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
            StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);
    
            // 1.4 基于Blink的批处理
            EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner()
                    .inBatchMode()
                    .build();
            TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);
    
            // 2. 表的创建：连接外部系统，读取数据
            // 2.1 读取文件
            String filePath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt";
            tableEnv.connect( new FileSystem().path(filePath))
                    .withFormat( new Csv())
                    .withSchema( new Schema()
                    .field("id", DataTypes.STRING())
                            .field("timestamp", DataTypes.BIGINT())
                            .field("temp", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("inputTable");
    
            Table inputTable = tableEnv.from("inputTable");
    //        inputTable.printSchema();
    //        tableEnv.toAppendStream(inputTable, Row.class).print();
    
            // 3. 查询转换
            // 3.1 Table API
            // 简单转换
            Table resultTable = inputTable.select("id, temp")
                    .filter("id === 'sensor_6'");
    
            // 聚合统计
            Table aggTable = inputTable.groupBy("id")
                    .select("id, id.count as count, temp.avg as avgTemp");
    
            // 3.2 SQL
            tableEnv.sqlQuery("select id, temp from inputTable where id = 'senosr_6'");
            Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id");
    
            // 打印输出
            tableEnv.toAppendStream(resultTable, Row.class).print("result");
            tableEnv.toRetractStream(aggTable, Row.class).print("agg");
            tableEnv.toRetractStream(sqlAggTable, Row.class).print("sqlagg");
    
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/f54fe529ae8a4639ba770b93260734bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/81f4833c38604a01a4f6a304371d7360.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/e2462691e6aa42839d22af7852cf5b04.png