星爷的博客

Flink(6)——flink table & sql 介绍

March 6, 2019
作者：星爷
出处：http://www.readingnotes.site/posts/Flink(6)-Flink-Table-&-SQL-%E4%BB%8B%E7%BB%8D.html
声明：转载请注明作者及出处。

本文基于 Flink 1.7。

随着 Hadoop 的发展，有了Hive，使用HQL 即可完成原来繁琐的Map Reduce 程序。

随着 Spark的发展，引入了 Spark SQL。

随着 Flink 版本的更迭，Flink 也提供了Flink SQL，以及 Table APIs。

注意：截止 Flink 1.7，Table API 和 SQL 还没有完成，有些操作还不支持。

1. 基本概念

1.1 Why

那么，为什么要推出Table APIs和SQL？

首先，来看下Flink 的编程模型。如下图所示（图片来源于官网），DataStream API 和 DataSet API 是分开的，但是对于应用开发者来说，为什么要关注这一点？对于相同的数据，批处理与流计算居然要写两套代码，简直不可思议。Table APIs和SQL的推出，实现了流处理和批处理的统一。

其次，降低了学习和使用门槛，基于 DataStream/DataSet APIs 的 Scala 或 Java 程序开发，对于BI/分析师来说，还是有一定门槛的，而SQL 则简单太多了。

1.2 Dynamic Tables

Dynamic Tables 是 Flink Table API 和 SQL的核心概念，与大家熟知的Static Tables 相比，Dynamic Tables 随着时间一直在变化。可以查询Dynamic Table，查询Dynamic Table 时会产生一个持续的查询，持续的查询不会终止，产生的结果也是Dynamic Table，根据输入，输出也会不断变化。熟悉关系型数据库的可以将Dynamic Tables的查询跟关系型数据库里查询物化视图对比起来，需要注意的是，Dynamic Tables 是一个逻辑概念，不需要（全部）物化。

另外，需要注意，在Dynamic Table上的持续查询的结果语义上是跟在Dynamic Table的快照上执行查询相同。

dynamic table & continuous query & stream

如上图所示：

1.3 Update Queries VS Append Queries

Append Queries：只会对查询结果进行追加的查询。

Update Queries：会更新查询结果的查询，一般需要维护更多的state。

1.4 查询限制

有些 Stream 上的查询需要花费巨大的代价：

1.5 Table 到 Stream 的转化

就像普通的数据库Table 一样，Dynamic Table也支持 insert、update、delete等对它的更新。当需要将Dynamic Table 转化为 Stream 或者输出到外部系统时，需要对这些更新进行encode。

Append-only Stream 和 Retarct Stream 支持将Dynamic Table 转化为DataStream。

2. 实战

下面引入一个简单的例子，从stream开始，转化为 Table，然后查询Table，最后将Table 转化为Stream。

从例子可以很容易的看出，Stream 和 Table APIs / SQL 可以很容易的混用，这也给我们带来了极大的便利性。

2.1 引入依赖

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-table_2.11</artifactId>
			<version>${flink.version}</version>
		</dependency>
		<dependency>
            <!-- for batch query -->
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-scala_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
			<!-- 上线时用provided，避免build的jar包太大，更避免冲突 -->
			<!--<scope>provided</scope>-->
			<!-- IDEA 里用compile，否则in-ide execution会失败 -->
			<scope>compile</scope>
		</dependency>
		<dependency>
            <!-- for streaming query -->
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
			<!-- 上线时用provided，避免build的jar包太大，更避免冲突 -->
			<!--<scope>provided</scope>-->
			<!-- IDEA 里用compile，否则in-ide execution会失败 -->
			<scope>compile</scope>
		</dependency>

2.2 隐式转换

Flink 的 Scala Table APIs用了隐式转换，所以，需要import 进来。

    import org.apache.flink.table.api.scala._
    import org.apache.flink.api.scala._

2.3 创建 TableEnvironment

TableEnvironment 是 Table APIs 和 SQL 的核心，可以用于：

Table 总是绑定到一个 TableEnvironment的，在使用时，在同一个SQL中不能联合使用不同TableEnvironment的表，比如join或union。

下面创建一个用于Stream的StreamTableEnvironment。另外，创建一个简单的stream。

    // 创建StreamTableEnvironment
    val senv = StreamExecutionEnvironment.getExecutionEnvironment
    val stableEnv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(senv)
  
    // 创建一个用于实验的 Stream[ObjPrice]
    case class ObjPrice (name: String, price: Long)
    val stream: DataStream[ObjPrice] = senv.fromCollection(List(ObjPrice("car", 100000), ObjPrice("house", 2000000), ObjPrice("book", 100), ObjPrice("car", 900210)))

2.4 将Stream 转化为 Table

    val sTable1Rename: Table = stableEnv.fromDataStream(stream, 'myName, 'myPrice)

将上面的stream 转化为 Table，同时对字段进行重命名。

2.5 查询 Table

    // 采用Table API 的方式进行查询
    val sTableResult: Table = sTable1Rename
      .filter('myPrice > 1000)
      .groupBy('myName)
      .select('myName, 'myPrice.sum as 'mySumPrice)

2.6 将 Table 转化为 Stream

    val sResultDataStream: DataStream[(Boolean, ObjPrice)] = stableEnv.toRetractStream[ObjPrice](sTableResult)

3. 总结

本文仅涉及一些基础知识和最常见的使用，其他的比如注册 Table / TableSink / TableSource / External Catalog 、数据类型与Table Schema的映射、查询优化等并不涉及，可以参考官网 进行查阅。

为了方便交流，请扫描下方二维码关注我。

QR