Table API&SQL的基本概念及使用介绍

 浪尖 浪尖聊大数据

Table API和SQL集成在共同API中。这个API的中心概念是一个用作查询的输入和输出的表。本文档显示了具有表API和SQL查询的程序的常见结构，如何注册表，如何查询表以及如何发出表。

Table API和SQL捆绑在flink-table Maven工程中。 为了使用Table API和SQL，必须将以下依赖项添加到您的项目中：

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-table_2.10</artifactId>
   <version>1.3.2</version>
</dependency>

此外，您需要为Flink的Scala批处理或流式API添加依赖关系。 对于批量查询，您需要添加：

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-scala_2.10</artifactId>
   <version>1.3.2</version>
</dependency>

对于流式查询，您需要添加：

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-streaming-scala_2.10</artifactId>
   <version>1.3.2</version>
</dependency>

注意：由于Apache Calcite中的一个问题，阻止用户类加载器被垃圾回收，我们不建议构建一个包含flink-table依赖项的fat-jar。相反，我们建议将Flink配置为在系统类加载器中包含flink-table依赖关系。这可以通过将./opt文件夹中的flink-table.jar文件复制到./lib文件夹来完成。

 

一，Table API&Sql项目的结构

用于批处理和流式处理的所有Table  API和SQL程序都遵循相同的模式。以下代码示例显示了Table API和SQL程序的通用结构。

// for batch programs use ExecutionEnvironment instead of StreamExecutionEnvironment
val env = StreamExecutionEnvironment.getExecutionEnvironment

// create a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// register a Table
tableEnv.registerTable("table1", ...)           // or
tableEnv.registerTableSource("table2", ...)     // or
tableEnv.registerExternalCatalog("extCat", ...)

// create a Table from a Table API query
val tapiResult = tableEnv.scan("table1").select(...)
// Create a Table from a SQL query
val sqlResult  = tableEnv.sql("SELECT ... FROM table2 ...")

// emit a Table API result Table to a TableSink, same for SQL result
tapiResult.writeToSink(...)

// execute
env.execute()

注意：Table API和SQL查询可以轻松集成到DataStream或DataSet程序中。

 

二，创建一个TableEnvironment

TableEnvironment是Table API和SQL集成的核心概念。 它负责：

A),在内部catalog中注册表

B),注册外部catalog

C),执行SQL查询

D),注册用户定义（标量，表或聚合）函数

E),将DataStream或DataSet转换为表

F),持有对ExecutionEnvironment或StreamExecutionEnvironment的引用

表总是绑定到一个特定的TableEnvironment。不可能在同一个查询中组合不同TableEnvironments的表，例如，join或union它们。通过使用StreamExecutionEnvironment或ExecutionEnvironment和可选的TableConfig调用静态TableEnvironment.getTableEnvironment()方法创建TableEnvironment。TableConfig可用于配置TableEnvironment或自定义查询优化和翻译过程（请参阅查询优化）。

// ***************
// STREAMING QUERY
// ***************
val sEnv = StreamExecutionEnvironment.getExecutionEnvironment
// create a TableEnvironment for streaming queries
val sTableEnv = TableEnvironment.getTableEnvironment(sEnv)

// ***********
// BATCH QUERY
// ***********
val bEnv = ExecutionEnvironment.getExecutionEnvironment
// create a TableEnvironment for batch queries
val bTableEnv = TableEnvironment.getTableEnvironment(bEnv)

 

三，在Catalog中注册一张表

TableEnvironment具有一个表的内部Catalog，按表名组织。 表API或SQL查询可以通过名称引用来访问Catalog中注册的表。

TableEnvironment允许您从各种来源注册表：

A),一个现有的Table对象，通常是一个Table API或SQL查询的结果。

B),一个TableSource，用于访问外部数据，如文件，数据库或消息系统。

C),来自DataStream或DataSet程序的DataStream或DataSet。

 

1,注册一张表

表在TableEnvironment中注册如下：

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// Table is the result of a simple projection query 
val projTable: Table = tableEnv.scan("X").project(...)

// register the Table projTable as table "projectedX"
tableEnv.registerTable("projectedTable", projTable)

注意：注册的表格与从关系数据库系统所知道的VIEW类似，即定义该表的查询未被优化，但是当另一个查询引用已注册的表时将被内联处理。如果多个查询引用相同的注册表，则每个引用查询将被内联并执行多次，即注册表的结果将不会被共享。

 

2，注册TableSource

TableSource提供对存储在诸如数据库（MySQL，HBase，...）的存储系统中的外部数据的访问，具有特定编码的文件（CSV，Apache [Parquet，Avro，ORC]，...））或消息系统 （Apache Kafka，RabbitMQ，...）。

Flink旨在为通用数据格式和存储系统提供TableSources。后面还会出文章讲解TablesSources和Sinks。

TableSource在TableEnvironment中注册如下：

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// create a TableSource
val csvSource: TableSource = new CsvTableSource("/path/to/file", ...)

// register the TableSource as table "CsvTable"
tableEnv.registerTableSource("CsvTable", csvSource)

 

四，注册一个外部Catalog

外部Catalog可以提供有关外部数据库和表的信息，例如其名称，模式，统计信息和有关如何访问存储在外部数据库，表或文件中的数据的信息。可以通过实现ExternalCatalog界面创建外部目录，并在TableEnvironment中注册如下：

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// create an external catalog
val catalog: ExternalCatalog = new InMemoryExternalCatalog

// register the ExternalCatalog catalog
tableEnv.registerExternalCatalog("InMemCatalog", catalog)

一旦注册到TableEnvironment中，可以通过指定其完整路径（如catalog.database.table）从Table API或SQL查询中访问ExternalCatalog中定义的所有表。目前，Flink为演示和测试提供了一个InMemoryExternalCatalog。但是，ExternalCatalog界面也可用于将目录（如HCatalog或Metastore）连接到Table API。

 

五，查询表

 

1，Table API

Table API是用于Scala和Java的语言集成查询API。与SQL相反，查询没有被指定为字符串，而是在主机语言中逐步构建。后面会出文章详细介绍这个特性。

该API基于Table类，代表一张表(Streaming或者batch),提供使用相关操作的方法。这些方法返回一个新的Table对象，它表示在输入表中应用关系操作的结果。一些关系操作由多个方法调用组成，如table.groupBy（...）.select（），其中groupBy（...）指定分组表，select（...） 从分组表中选取结果。

以下示例显示了一个简单的Table API聚合查询：

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// register Orders table

// scan registered Orders table
Table orders = tableEnv.scan("Orders")
// compute revenue for all customers from France
Table revenue = orders
  .filter('cCountry === "FRANCE")
  .groupBy('cID, 'cName)
  .select('cID, 'cName, 'revenue.sum AS 'revSum)

注意：Scala Table API使用Scala符号，它以单个tick（'）开始引用表的属性。Table API使用Scala隐含。 确保导入org.apache.flink.api.scala._和org.apache.flink.table.api.scala._以便使用Scala隐式转换。

 

2，SQL

Flink的SQL集成基于实现SQL标准的Apache Calcite。 SQL查询被指定为常规字符串。后面会出文章详细介绍这个特性。

 

以下示例显示如何指定查询并将结果作为表返回。

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// register Orders table

// compute revenue for all customers from France
Table revenue = tableEnv.sql(""" 
                               |SELECT cID, cName, SUM(revenue) AS revSum
                               |FROM Orders
                               |WHERE cCountry = 'FRANCE'
                               |GROUP BY cID, cName
                             """.stripMargin)

// emit or convert Table
// execute query

 

3，混合Table API和SQL

Table API查询可以在SQL查询返回的Table对象上进行操作。

通过将Table API返回的对象注册成表也可以进行一个SQL查询请求，在SQL查询的FROM子句中引用它。

 

六，输出一张表

为了输出一个表，可以将它写入一个TableSink。TableSink是支持各种文件格式（例如CSV，Apache Parquet，Apache Avro），存储系统（例如JDBC，Apache HBase，Apache Cassandra，Elasticsearch）或消息传递系统（例如Apache Kafka，RabbitMQ的）。

批处理表只能写入BatchTableSink，而streaming table需要一个AppendStreamTableSink，一个RetractStreamTableSink或一个UpsertStreamTableSink。

有关Table source和sink的详细信息及如何自定义一个TableSink后面会给出详细的文章。使用例子：

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// compute a result Table using Table API operators and/or SQL queries
val result: Table = ...

// create a TableSink
val sink: TableSink = new CsvTableSink("/path/to/file", fieldDelim ="|")

// write the result Table to the TableSink
result.writeToSink(sink)

// execute the program

 

七，翻译并执行查询

Table API和SQL查询根据输入是流式还是批量输入而被转换为DataStream或DataSet程序。 查询内部表示为逻辑查询计划，并分为两个阶段：

A),优化逻辑计划

B),翻译成DataStream或DataSet程序。

Table API或者SQL查询在一下情况下被翻译：

A),表被输出到TableSink，即当调用Table.writeToSink()时。

B),该表被转化为DataStream或者DataSet。

转换结束之后，Table API或SQL查询就像常规DataStream或DataSet程序一样处理，并在调用StreamExecutionEnvironment.execute()或ExecutionEnvironment.execute()时执行。

 

八，与DataStream和DataSet API集成

Table API和SQL查询可以轻松地集成到DataStream和DataSet程序中并嵌入到其中。表API和SQL查询可以轻松地集成到DataStream和DataSet程序中并嵌入到其中。 例如，可以查询外部表（例如来自RDBMS），进行一些预处理，例如过滤，投影，聚合或与元数据连接，然后使用DataStream或 DataSet API（以及任何构建在这些API之上的库，如CEP或Gelly）。 相反，Table API或SQL查询也可以应用于DataStream或DataSet程序的结果。相反，Table API或SQL查询也可以应用于DataStream或DataSet程序的结果。

这种交互可以通过将DataStream或DataSet转换为Table来实现，反之亦然。在本节中，我们将介绍如何完成这些转换。

 

1，Scala的隐式转换

Scala Table API提供DataSet，DataStream和Table类的隐式转换。通过导入包org.apache.flink.table.api.scala._除了用于Scala DataStream API的org.apache.flink.api.scala._之外还可以启用这些转换。

 

2，将DataStream或DataSet注册为表

结果表的schema 取决于注册的DataStream或DataSet的数据类型。有关详细信息，请查看有关将数据类型映射到表模式的部分。

// get TableEnvironment 
// registration of a DataSet is equivalent
val tableEnv = TableEnvironment.getTableEnvironment(env)

val stream: DataStream[(Long, String)] = ...

// register the DataStream as Table "myTable" with fields "f0", "f1"
tableEnv.registerDataStream("myTable", stream)

// register the DataStream as table "myTable2" with fields "myLong", "myString"
tableEnv.registerDataStream("myTable2", stream, 'myLong, 'myString)

 

3，将DataStream或DataSet转换为表

不仅仅可以在TableEnvironment中注册DataStream或DataSet，也可以直接转换为Table。 如果要在Table API查询中使用Table，这很方便。

// get TableEnvironment
// registration of a DataSet is equivalent
val tableEnv = TableEnvironment.getTableEnvironment(env)

val stream: DataStream[(Long, String)] = ...

// convert the DataStream into a Table with default fields '_1, '_2
val table1: Table = tableEnv.fromDataStream(stream)

// convert the DataStream into a Table with fields 'myLong, 'myString
val table2: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)

 

4，将表转换为DataStream或DataSet

表可以转换为DataStream或DataSet。以这种方式，可以基于Table API或SQL查询的结果运行自定义DataStream或DataSet程序。

将表转换为DataStream或DataSet时，需要指定生成的DataStream或DataSet的数据类型，即要转换表的行的数据类型。通常最方便的转换类型是Row。以下列表概述了不同选项的功能：

Row：字段通过位置，任意数量的字段映射，支持空值，无类型安全访问。

POJO：按名称映射字段（POJO字段必须命名为表字段），任意字段数，支持空值，类型安全访问。

Case Class：字段按位置映射，不支持空值，类型安全访问。

Tuple：字段通过位置映射，限制为22（Scala）或25（Java）字段，不支持空值，类型安全访问。

Atomic Type：表必须有单个字段，不支持空值，类型安全访问。

 

4.1 将表转换为DataStream

作为流式查询的结果的表将被动态地更新，即当新记录到达查询的输入流时，它会改变。因此，转换此动态查询的DataStream需要对表的更新进行编码。

将Table转换为DataStream有两种模式：

Append Mode：仅当动态表仅由INSERT更改修改时，才能使用此模式，即只是附加的，并且以前发布的结果永远不会被更新。

Retract Mode：始终可以使用此模式。 它使用布尔标志来编码INSERT和DELETE更改。

// get TableEnvironment. 
// registration of a DataSet is equivalent
val tableEnv = TableEnvironment.getTableEnvironment(env)

// Table with two fields (String name, Integer age)
val table: Table = ...

// convert the Table into an append DataStream of Row
val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)

// convert the Table into an append DataStream of Tuple2[String, Int]
val dsTuple: DataStream[(String, Int)] dsTuple =
  tableEnv.toAppendStream[(String, Int)](table)

// convert the Table into a retract DataStream of Row.
//   A retract stream of type X is a DataStream[(Boolean, X)]. 
//   The boolean field indicates the type of the change. 
//   True is INSERT, false is DELETE.
val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)

注意:有关动态表及其配置属性的介绍后面会出文章。

 

4.2 将Table转化为DataSet

将表转换为DataSet，如下所示：

// get TableEnvironment 
// registration of a DataSet is equivalent
val tableEnv = TableEnvironment.getTableEnvironment(env)

// Table with two fields (String name, Integer age)
val table: Table = ...

// convert the Table into a DataSet of Row
val dsRow: DataSet[Row] = tableEnv.toDataSet[Row](table)

// convert the Table into a DataSet of Tuple2[String, Int]
val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet[(String, Int)](table)

 

九，数据类型和表schema映射

Flink的DataStream和DataSet API支持非常多样化的类型，例如Tuples（内置Scala和Flink Java元组），POJO，Case Class和原子类型。下面我们将介绍Table API如何将这些类型转换为内部行表示，并显示将DataStream转换为Table的示例。

 

1，原子类型

Flink将原始（Integer，Double，String）或通用类型（无法分析和分解的类型）视为原子类型。属性的类型是从原子类型推断的，必须指定属性的名称。

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

val stream: DataStream[Long] = ...
// convert DataStream into Table with field 'myLong
val table: Table = tableEnv.fromDataStream(stream, 'myLong)

 

2，元组（Scala和Java）和Case Class（仅限Scala）

Flink支持Scala的内置元组，并为Java提供自己的元组类。两种元组的DataStreams和DataSet可以转换成表。可以通过为所有字段提供名称（基于位置的映射）来重命名字段。如果未指定字段名称，则使用默认字段名称。

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

val stream: DataStream[(Long, String)] = ...

// convert DataStream into Table with field names 'myLong, 'myString
val table1: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)

// convert DataStream into Table with default field names '_1, '_2
val table2: Table = tableEnv.fromDataStream(stream)

// define case class
case class Person(name: String, age: Int)
val streamCC: DataStream[Person] = ...

// convert DataStream into Table with default field names 'name, 'age
val tableCC1 = tableEnv.fromDataStream(streamCC)

// convert DataStream into Table with field names 'myName, 'myAge
val tableCC1 = tableEnv.fromDataStream(streamCC, 'myName, 'myAge)

 

3，POJO (Java and Scala)

Flink支持POJO作为复合类型。在这里记录了确定POJO的规则。将POJO DataStream或DataSet转换为Table而不指定字段名称时，将使用原始POJO字段的名称。重命名原始POJO字段需要关键字AS，因为POJO字段没有固有的顺序。名称映射需要原始名称，不能通过位置来完成。

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// Person is a POJO with field names "name" and "age"
val stream: DataStream[Person] = ...

// convert DataStream into Table with field names 'name, 'age
val table1: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with field names 'myName, 'myAge
val table2: Table = tableEnv.fromDataStream(stream, 'name as 'myName,'age as 'myAge)

 

4，Row

Row数据类型支持任意数量的具有空值的字段和字段。字段名称可以通过RowTypeInfo指定，也可以将Row DataStream或DataSet转换为Table（基于位置）。

// get a TableEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
val stream: DataStream[Row] = ...

// convert DataStream into Table with field names 'name, 'age
val table1: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with field names 'myName, 'myAge
val table2: Table = tableEnv.fromDataStream(stream, 'myName, 'myAge)

 

十，查询优化

Apache Flink利用Apache Calcite来优化和翻译查询。目前执行的优化包括投影和过滤器下推，子查询去相关等各种查询重写。Flink还没有优化连接的顺序，而是按照查询中定义的顺序执行它们（FROM子句中的表的顺序和/或WHERE子句中的连接谓词的顺序）。

可以通过提供一个CalciteConfig对象来调整在不同阶段应用的优化规则集。这可以通过调用CalciteConfig.createBuilder())通过构建器创建，并通过调用tableEnv.getConfig.setCalciteConfig(calciteConfig)提供给TableEnvironment。

Table API提供了一种解释计算表的逻辑和优化查询计划的机制。这通过TableEnvironment.explain(table)方法完成。它返回一个描述三个计划的字符串：

1,关系查询的抽象语法树，即未优化的逻辑查询计划，

2,优化的逻辑查询计划

3,物理执行计划

以下代码显示了一个示例和相应的输出：

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env)

val table1 = env.fromElements((1, "hello")).toTable(tEnv, 'count,'word)
val table2 = env.fromElements((1, "hello")).toTable(tEnv, 'count,'word)
val table = table1
  .where('word.like("F%"))
  .unionAll(table2)

val explanation: String = tEnv.explain(table)
println(explanation)

执行计划如下：