赞
踩
insert into target
select a, count(b) as cnt
from Orders
group by a
注意:无论输入是连续(流处理)还是有界(批处理),在 Table 和 SQL 任一 API 中同一条查询语句是具有相同的语义并且会产出相同的结果的。这就是说为什么 Flink SQL 和 Table API 可以做到在用户接口层面的流批统一。xdm,用一套 SQL 既能跑流任务,也能跑批任务,它不香嘛?
Table API 和 SQL API 也与 DataStream API 做到了无缝集成。可以轻松地在三种 API 之间灵活切换。例如,可以使用 SQL 的 MATCH_RECOGNIZE
子句匹配出异常的数据,然后使用再转为 DataStream API 去灵活的构建针对于异常数据的自定义报警机制。
在大致了解了这两个 API 是干啥的之后,我们就可以直接来看看,怎么使用这两个 API 了。
根据小伙伴们使用的编程语言的不同(Java 或 Scala),需要将对应的依赖包添加到项目中。
Java 依赖如下:
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge_2.11</artifactId> <version>1.13.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.11</artifactId> <version>1.13.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.11</artifactId> <version>1.13.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>1.13.5</version> </dependency>
Scala 依赖如下:
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-scala-bridge_2.11</artifactId> <version>1.13.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.11</artifactId> <version>1.13.5</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.11</artifactId> <version>1.13.5</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>1.13.5</version> </dependency>
引入上述依赖之后,小伙伴萌就可以开始使用 Table / SQL API 了。具体案例如下文所示。
在小伙伴萌看下文之前,先看一下整体的思路,跟着博主思路走,会更清晰:
// 创建一个 TableEnvironment,为后续使用 SQL 或者 Table API 提供上线 EnvironmentSettings settings = EnvironmentSettings .newInstance() .inStreamingMode() // 声明为流任务 //.inBatchMode() // 声明为批任务 .build(); TableEnvironment tEnv = TableEnvironment.create(settings); // 创建一个输入表 tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )"); // 创建一个输出表 tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )"); // 1. 使用 Table API 做一个查询并返回 Table Table table2 = tableEnv.from("table1").select(...); // 2. 使用 SQl API 做一个查询并返回 Table Table table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ... "); // 将 table2 的结果使用 Table API 写入 outputTable 中,并返回结果 TableResult tableResult = table2.executeInsert("outputTable"); tableResult...
总结一下上面案例使用到的一些 API,让大家先对 Table / SQL API 的能力有一个大概了解:
TableEnvironment
:Table API 和 SQL API 的都集成在一个 统一上下文(即 TableEnvironment
)中,其地位等同于 DataStream API 中的 StreamExecutionEnvironment
的地位TableEnvironment::executeSql
:用于 SQL API 中,可以执行一段完整 DDL、DML SQL。举例,方法入参可以是 CREATE TABLE xxx
,INSERT INTO xxx SELECT xxx FROM xxx
。TableEnvironment::from(xxx)
:用于 Table API 中,可以以强类型接口的方式运行。方法入参是一个表名称。TableEnvironment::sqlQuery
:用于 SQL API 中,可以执行一段查询 SQL,并把结果以 Table 的形式返回。举例,方法的入参是 SELECT xxx FROM xxx
。Table::executeInsert
:用于将 Table 的结果插入到结果表中。方法入参是写入的目标表。无论是对于 SQL API 来说还是对于 Table API 来说,都是使用 TableEnvironment
接口承载我们的业务查询逻辑的。只是在用户的使用接口的方式上有区别,以上述的 Java 代码为例,Table API 其实就是模拟 SQL 的查询方式封装了 Java 语言的 lambda 强类型 API,SQL 就是纯 SQL 查询。Table 和 SQL 很多时候都是掺杂在一起的,大家理解的时候就可以直接将 Table 和 SQL API 直接按照 SQL 进行理解,不用强行做特殊的区分。
而且博主推荐的话,直接上 SQL API 就行,其实 Table API 在企业实战中用的不是特别多。你说 Table API 方便吧,它确实比 DataStream API 方便,但是又比 SQL 复杂。一般生产使用不多。
注意:由于 Table 和 SQL API 基本上属于一回事,后续如果没有特别介绍的话,博主就直接按照 SQL API 进行介绍了。
TableEnvironment
是使用 SQL API 永远都离不开的一个接口。其是 SQL API 使用的入口(上下文),就像是你要使用 Java DataStream API 去写一个 Flink 任务需要使用到 StreamExecutionEnvironment
一样。
可以认为 TableEnvironment
在 SQL API 中的地位和 StreamExecutionEnvironment
在 DataStream 中的地位是一样的,都是包含了一个 Flink 任务运行时的所有上下文环境信息。大家这样对比学习会比较好理解。
TableEnvironment
包含的功能如下:
Catalog
可以理解为 Flink 的 MetaStore
,类似 Hive MetaStore
对在 Hive 中的地位,关于 Flink Catalog 的详细内容后续进行介绍。addSource
、map
、flatmap
等接口。DataStream 和 Table(Table / SQL API 的查询结果)之间进行转换:目前
1.13
1.13
1.13 版本的只有流任务支持,批任务不支持。
1.14
1.14
1.14 支持批任务。
接下来介绍如何创建一个 TableEnvironment
。案例为 Java。
EnvironmentSettings
创建 TableEnvironment
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
// 1. 就是设置一些环境信息
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.inStreamingMode() // 声明为流任务
//.inBatchMode() // 声明为批任务
.build();
// 2. 创建 TableEnvironment
TableEnvironment tEnv = TableEnvironment.create(settings);
在
1.13
1.13
1.13 版本中:
inStreamingMode
,则最终创建出来的 TableEnvironment
实例为 StreamTableEnvironmentImpl
。inBatchMode
,则最终创建出来的 TableEnvironment
实例为 TableEnvironmentImpl
。它两虽然都继承了 TableEnvironment
接口,但是 StreamTableEnvironmentImpl
支持的功能更多一些。大家可以直接去看看接口实验一下,这里就不进行详细介绍。
StreamExecutionEnvironment
创建 TableEnvironment
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
一个表的全名(标识)会由三个部分组成:Catalog 名称.数据库名称.表名称。如果 Catalog 名称或者数据库名称没有指明,就会使用当前默认值 default
。
举个例子,下面这个 SQL 创建的 Table 的全名为 default.default.table1
。
tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");
下面这个 SQL 创建的 Table 的全名为 default.mydatabase.table1
。
tableEnv.executeSql("CREATE TEMPORARY TABLE mydatabase.table1 ... WITH ( 'connector' = ... )");
表 可以是 常规的(外部表 TABLE),也可以是 虚拟的(视图 VIEW)。
注意:这里有不同的地方就是,离线 Hive MetaStore 中不会有 Catalog 这个概念,其标识都是 数据库.数据表。
临时表:通常保存于内存中并且仅在创建它们的 Flink Session(可以理解为一次 Flink 任务的运行)持续期间存在。这些表对于其它 Session(即其他 Flink 任务或非此次运行的 Flink 任务)是不可见的。因为这个表的元数据没有被持久化。如下案例:
-- 临时外部表
CREATE TEMPORARY TABLE source_table (
user_id BIGINT,
`name` STRING
) WITH (
'connector' = 'user\_defined',
'format' = 'json',
'class.name' = 'flink.examples.sql.\_03.source\_sink.table.user\_defined.UserDefinedSource'
);
-- 临时视图
CREATE TEMPORARY VIEW query_view as
SELECT \*
FROM source_table;
永久表:需要外部 Catalog(例如 Hive Metastore)来持久化表的元数据。一旦永久表被创建,它将对任何连接到这个 Catalog 的 Flink Session 可见且持续存在,直至从 Catalog 中被明确删除。如下案例:
-- 永久外部表。需要外部 Catalog 持久化!!!
CREATE TABLE source_table (
user_id BIGINT,
`name` STRING
) WITH (
'connector' = 'user\_defined',
'format' = 'json',
'class.name' = 'flink.examples.sql.\_03.source\_sink.table.user\_defined.UserDefinedSource'
);
-- 永久视图。需要外部 Catalog 持久化!!!
CREATE VIEW query_view as
SELECT \*
FROM source_table;
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/小蓝xlanll/article/detail/659339
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。