Flink Table API & SQL编程指南(1)_flink-table-runtime

Apache Flink提供了两种顶层的关系型API，分别为Table API和SQL，Flink通过Table API&SQL实现了批流统一。其中Table API是用于Scala和Java的语言集成查询API，它允许以非常直观的方式组合关系运算符（例如select，where和join）的查询。Flink SQL基于Apache Calcite 实现了标准的SQL，用户可以使用标准的SQL处理数据集。Table API和SQL与Flink的DataStream和DataSet API紧密集成在一起，用户可以实现相互转化，比如可以将DataStream或者DataSet注册为table进行操作数据。值得注意的是，Table API and SQL目前尚未完全完善，还在积极的开发中，所以并不是所有的算子操作都可以通过其实现。

依赖

从Flink1.9开始，Flink为Table & SQL API提供了两种planner,分别为Blink planner和old planner，其中old planner是在Flink1.9之前的版本使用。主要区别如下：

尖叫提示：对于生产环境，目前推荐使用old planner.

• flink-table-common: 通用模块，包含 Flink Planner 和 Blink Planner 一些共用的代码
• flink-table-api-java: java语言的Table & SQL API，仅针对table(处于早期的开发阶段，不推荐使用)
• flink-table-api-scala: scala语言的Table & SQL API，仅针对table(处于早期的开发阶段，不推荐使用)
• flink-table-api-java-bridge: java语言的Table & SQL API，支持DataStream/DataSet API(推荐使用)
• flink-table-api-scala-bridge: scala语言的Table & SQL API，支持DataStream/DataSet API(推荐使用)
• flink-table-planner:planner 和runtime. planner为Flink1,9之前的old planner(推荐使用)
• flink-table-planner-blink: 新的Blink planner.
• flink-table-runtime-blink: 新的Blink runtime.
• flink-table-uber: 将上述的API模块及old planner打成一个jar包，形如flink-table-*.jar，位与/lib目录下
• flink-table-uber-blink:将上述的API模块及Blink 模块打成一个jar包，形如fflink-table-blink-*.jar，位与/lib目录下

Blink planner & old planner

Blink planner和old planner有许多不同的特点，具体列举如下：

• Blink planner将批处理作业看做是流处理作业的特例。所以，不支持Table 与DataSet之间的转换，批处理的作业也不会被转成DataSet程序，而是被转为DataStream程序。
• Blink planner不支持 BatchTableSource，使用的是有界的StreamTableSource。
• Blink planner仅支持新的 Catalog，不支持ExternalCatalog (已过时)。
• 对于FilterableTableSource的实现，两种Planner是不同的。old planner会谓词下推到PlannerExpression(未来会被移除)，而Blink planner 会谓词下推到 Expression(表示一个产生计算结果的逻辑树)。
• 仅仅Blink planner支持key-value形式的配置，即通过Configuration进行参数设置。
• 关于PlannerConfig的实现，两种planner有所不同。
• Blink planner 会将多个sink优化成一个DAG(仅支持TableEnvironment，StreamTableEnvironment不支持)，old planner总是将每一个sink优化成一个新的DAG，每一个DAG都是相互独立的。
• old planner不支持catalog统计，Blink planner支持catalog统计。

Flink Table & SQL程序的pom依赖

根据使用的语言不同，可以选择下面的依赖，包括scala版和java版，如下：

<!-- java版 -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge_2.11</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
<!-- scala版 -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

除此之外，如果需要在本地的IDE中运行Table API & SQL的程序，则需要添加下面的pom依赖：

<!-- Flink 1.9之前的old planner -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-planner_2.11</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
<!-- 新的Blink planner -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-planner-blink_2.11</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

另外，如果需要实现自定义的格式(比如和kafka交互)或者用户自定义函数，需要添加如下依赖：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-common</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
1
2
3
4
5
6

Table API & SQL的编程模板

所有的Table API&SQL的程序(无论是批处理还是流处理)都有着相同的形式，下面将给出通用的编程结构形式：

// 创建一个TableEnvironment对象，指定planner、处理模式(batch、streaming)
TableEnvironment tableEnv = ...; 
// 创建一个表
tableEnv.connect(...).createTemporaryTable("table1");
// 注册一个外部的表
tableEnv.connect(...).createTemporaryTable("outputTable");
// 通过Table API的查询创建一个Table 对象
Table tapiResult = tableEnv.from("table1").select(...);
// 通过SQL查询的查询创建一个Table 对象
Table sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");
// 将结果写入TableSink
tapiResult.insertInto("outputTable");
// 执行
tableEnv.execute("java_job");
1
2
3
4
5
6
7
8
9
10
11
12
13
14

注意：Table API & SQL的查询可以相互集成，另外还可以在DataStream或者DataSet中使用Table API & SQL的API，实现DataStreams、 DataSet与Table之间的相互转换。

创建TableEnvironment

TableEnvironment是Table API & SQL程序的一个入口，主要包括如下的功能：

• 在内部的catalog中注册Table
• 注册catalog
• 加载可插拔模块
• 执行SQL查询
• 注册用户定义函数
• DataStream 、DataSet与Table之间的相互转换
• 持有对ExecutionEnvironment 、StreamExecutionEnvironment的引用

一个Table必定属于一个具体的TableEnvironment，不可以将不同TableEnvironment的表放在一起使用(比如join，union等操作)。

TableEnvironment是通过调用 Bat