详解 Flink Table API 和 Flink SQL 之入门介绍

一、介绍

• Flink 提供了对于“表”处理的支持，这就是更高层级的应用 API，在 Flink 中被称为 Table API 和 SQL
• Table API 是基于“表”（Table）的一套 API，它是内嵌在 Java、Scala 等语言中的一种声明式领域特定语言（DSL），也就是专门为处理表而设计的
• Flink SQL 是基于 Apache Calcite 实现了对 SQL 的支持
• Table API 和 SQL 最初并不完善，在 Flink 1.9 版本合并阿里巴巴内部版本 Blink 之后发生了非常大的改变，此后也一直处在快速开发和完善的过程中，直到 Flink 1.12 版本才基本上做到了功能上的完善。而即使是在目前最新的 1.13 版本中， Table API 和 SQL 也依然不算稳定，接口用法还在不停调整和更新

二、快速入门

1. 引入依赖

<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>

2. 案例

public class TestTableAPIAndSql {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//读取数据并转换成 POJO
DataStream<String> inputStream = env.readTextFile("./sensor.txt");
DataStream<SensorReading> dataStream = inputStream.map(line -> {
String[] fields = line.split(",");
return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
});
//1. 创建 TableAPI 执行环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//2. 基于 DataStream 创建表
Table dataTable = tableEnv.fromDataStream(dataStream);
//3. 调用 TableAPI
Table resultTable = dataTable.select("id, temperature").where("id = 'sensor_1'");
//4. 使用 Flink SQL 语法
//4.1 基于 Table 注册一个查询视图
tableEnv.createTemporaryView("sensor", dataTable);
//4.2 对注册的视图进行 SQL 操作
String sql = "select id, temperature from sensor where id = 'sensor_1'";
Table resultSqlTable = tableEnv.sqlQuery(sql);
//5.打印输出
tableEnv.toAppendStream(resultTable, Row.class).print("table");
tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");
env.execute();
}
}

三、程序结构

TableAPI 和 SQL 程序的整体结构与 DataStream API 非常相似，也可以分为读取数据源（Source）、转换（Transform）和输出数据（Sink）三部分，TableAPI 的输入输出操作不需要额外定义，只需要将用于输入和输出的表定义出来，然后进行转换查询就可以

//1.创建表的执行环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);  
//2.创建一张表，用于读取数据源
tableEnv.connect(...).createTemporaryTable("inputTable");
//3.通过 TableAPI或SQL进行转换查询
Table result = tableEnv.from("inputTable").select(...);
Table sqlResult = tableEnv.sqlQuery("SELECT ... FROM inputTable ...");
//4.注册一张表，用于输出结果
tableEnv.connect(...).createTemporaryTable("outputTable");
result.insertInto("outputTable");