生鲜网站建设规划书,公司网站设计很好的,食品企业网站建设策划方案书,北京制作页面Flink SQL 基础概念#xff1a;SQL Table 运行环境、基本概念及常用 API 1.SQL Table 简介及运行环境1.1 简介1.2 SQL 和 Table API 运行环境依赖 2.SQL Table 的基本概念及常用 API2.1 一个 SQL / Table API 任务的代码结构2.2 SQL 上下文#xff1a;Tabl… Flink SQL 基础概念SQL Table 运行环境、基本概念及常用 API 1.SQL Table 简介及运行环境1.1 简介1.2 SQL 和 Table API 运行环境依赖 2.SQL Table 的基本概念及常用 API2.1 一个 SQL / Table API 任务的代码结构2.2 SQL 上下文TableEnvironment2.3 SQL 中表的概念外部表 TABLE、视图 VIEW2.4 SQL 临时表、永久表2.5 SQL 外部数据表2.5.1 Table API 创建外部数据表2.5.2 SQL API 创建外部数据表 2.6 SQL 视图 VIEW2.6.1 Table API 创建 VIEW2.6.2 SQL API 创建 VIEW 2.7 一个 SQL 查询案例2.8 SQL 与 DataStream API 的转换 1.SQL Table 简介及运行环境
1.1 简介
Apache Flink 提供了两种关系型 API 用于统一流和批处理Table 和 SQL API。
Table API 是一种集成在 Java、Scala 和 Python 语言中的查询 API简单理解就是用 Java、Scala、Python 按照 SQL 的查询接口封装了一层 lambda 表达式的查询 API它允许以强类型接口的方式组合各种关系运算符如选择、筛选和联接的查询操作然后生成一个 Flink 任务运行。如下案例所示
import org.apache.flink.table.api.*;import static org.apache.flink.table.api.Expressions.*;EnvironmentSettings settings EnvironmentSettings.newInstance().inStreamingMode().build();TableEnvironment tEnv TableEnvironment.create(settings);// 下面就是 Table API 的案例其语义等同于
// select a, count(b) as cnt
// from Orders
// group by a
DataSetRow result tEnv.from(Orders).groupBy($(a)).select($(a), $(b).count().as(cnt)).toDataSet(counts, Row.class);result.print();SQL API 是基于 SQL 标准的 Apache Calcite 框架实现的我们可以使用纯 SQL 来开发和运行一个 Flink 任务。如下案例所示
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 了。
1.2 SQL 和 Table API 运行环境依赖
根据小伙伴们使用的编程语言的不同Java 或 Scala需要将对应的依赖包添加到项目中。
Java 依赖如下
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-api-java-bridge_2.11/artifactIdversion1.13.5/version
/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-planner-blink_2.11/artifactIdversion1.13.5/version
/dependency
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java_2.11/artifactIdversion1.13.5/version
/dependency
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-common/artifactIdversion1.13.5/version
/dependencyScala 依赖如下
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-api-scala-bridge_2.11/artifactIdversion1.13.5/version
/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-planner-blink_2.11/artifactIdversion1.13.5/versionscopeprovided/scope
/dependency
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_2.11/artifactIdversion1.13.5/versionscopeprovided/scope
/dependency
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-common/artifactIdversion1.13.5/version
/dependency引入上述依赖之后小伙伴萌就可以开始使用 Table / SQL API 了。具体案例如下文所示。
2.SQL Table 的基本概念及常用 API
在小伙伴萌看下文之前先看一下整体的思路跟着博主思路走会更清晰
先通过一个 SQL / Table API 任务看一下我们在实际开发时的代码结构应该长啥样让大家能有直观的感受。重点介绍 SQL / Table API 中核心 API - TableEnvironment。SQL / Table 所有能用的接口都在 TableEnvironment 中。通过两个角度外部表 / 视图、临时 / 非临时认识 Flink SQL 体系中的表的概念。举几个创建外部表、视图的实际应用案例。
2.1 一个 SQL / Table 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 的能力有一个大概了解
TableEnvironmentTable API 和 SQL API 的都集成在一个 统一上下文即 TableEnvironment中其地位等同于 DataStream API 中的 StreamExecutionEnvironment 的地位TableEnvironment::executeSql用于 SQL API 中可以执行一段完整 DDL、DML SQL。举例方法入参可以是 CREATE TABLE xxxINSERT 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 强类型 APISQL 就是纯 SQL 查询。Table 和 SQL 很多时候都是掺杂在一起的大家理解的时候就可以直接将 Table 和 SQL API 直接按照 SQL 进行理解不用强行做特殊的区分。
而且博主推荐的话直接上 SQL API 就行其实 Table API 在企业实战中用的不是特别多。你说 Table API 方便吧它确实比 DataStream API 方便但是又比 SQL 复杂。一般生产使用不多。
注意由于 Table 和 SQL API 基本上属于一回事后续如果没有特别介绍的话博主就直接按照 SQL API 进行介绍了。
2.2 SQL 上下文TableEnvironment
TableEnvironment 是使用 SQL API 永远都离不开的一个接口。其是 SQL API 使用的入口上下文就像是你要使用 Java DataStream API 去写一个 Flink 任务需要使用到 StreamExecutionEnvironment 一样。
可以认为 TableEnvironment 在 SQL API 中的地位和 StreamExecutionEnvironment 在 DataStream 中的地位是一样的都是包含了一个 Flink 任务运行时的所有上下文环境信息。大家这样对比学习会比较好理解。
TableEnvironment 包含的功能如下
Catalog 管理Catalog 可以理解为 Flink 的 MetaStore类似 Hive MetaStore 对在 Hive 中的地位关于 Flink Catalog 的详细内容后续进行介绍。表管理在 Catalog 中注册表。SQL 查询这 TMD 还用说最基本的功能啊就像 DataStream 中提供了 addSource、map、flatmap 等接口。UDF 管理注册用户定义标量函数一进一出、表函数一进多出、聚合函数多进一出函数。UDF 扩展加载可插拔 ModuleModule 可以理解为 Flink 管理 UDF 的模块是可插拔的可以让小伙伴萌自定义 Module去支持奇奇怪怪的 UDF 功能。
DataStream 和 TableTable / SQL API 的查询结果之间进行转换目前 1.13 1.13 1.13 版本的只有流任务支持批任务不支持。 1.14 1.14 1.14 支持批任务。
接下来介绍如何创建一个 TableEnvironment。案例为 Java。
方法 1通过 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 支持的功能更多一些。大家可以直接去看看接口实验一下这里就不进行详细介绍。
方法 2通过已有的 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);2.3 SQL 中表的概念外部表 TABLE、视图 VIEW
一个表的全名标识会由三个部分组成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。
外部表 TABLE描述的是外部数据例如文件HDFS、消息队列Kafka等。依然拿离线 Hive SQL 举个例子离线中一个表指的是 Hive 表也就是所说的外部数据。视图 VIEW从已经存在的表中创建视图一般是一个 SQL 逻辑的查询结果。对比到离线的 Hive SQL 中在离线的场景Hive 表中 VIEW 也都是从已有的表中去创建的。其实 Flink 也是一样的。
注意这里有不同的地方就是离线 Hive MetaStore 中不会有 Catalog 这个概念其标识都是 数据库.数据表。
2.4 SQL 临时表、永久表
表视图、外部表可以是 临时的并与单个 Flink Session可以理解为 Flink 任务运行一次就是一个 Session的生命周期绑定。表视图、外部表也可以是 永久的并且对多个 Flink Session 都生效。
临时表通常保存于内存中并且仅在创建它们的 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;注意如果临时表和永久表使用了相同的名称Catalog名.数据库名.表名。那么在这个 Flink Session 中你的任务访问到这个表时访问到的永远是临时表即 相同名称的表临时表会屏蔽永久表。 2.5 SQL 外部数据表
由于目前在实时数据的场景中多以消息队列作为数据表。此处就以 Kafka 为例创建一个外部数据表。
2.5.1 Table API 创建外部数据表
public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());EnvironmentSettings settings EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tEnv StreamTableEnvironment.create(env, settings);// kafka 数据源DataStreamRow r env.addSource(new FlinkKafkaConsumerRow(xxx));// 将 DataStream 转为一个 Table API 中的 Table 对象进行使用Table sourceTable tEnv.fromDataStream(r, Schema.newBuilder().column(f0, string).column(f1, string).column(f2, bigint).columnByExpression(proctime, PROCTIME()).build());tEnv.createTemporaryView(source_table, sourceTable);String selectWhereSql select f0 from source_table where f1 b;Table resultTable tEnv.sqlQuery(selectWhereSql);tEnv.toRetractStream(resultTable, Row.class).print();env.execute();
}上述案例中Table API 将一个 DataStream 的结果集通过 StreamTableEnvironment::fromDataStream 转为一个 Table 对象来使用。
2.5.2 SQL API 创建外部数据表
EnvironmentSettings settings EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tEnv StreamTableEnvironment.create(env, settings);// SQL API 执行 create table 创建表
tEnv.executeSql(CREATE TABLE KafkaSourceTable (\n f0 STRING,\n f1 STRING\n ) WITH (\n connector kafka,\n topic topic,\n properties.bootstrap.servers localhost:9092,\n properties.group.id testGroup,\n format json\n )
);Table t tEnv.sqlQuery(SELECT * FROM KafkaSourceTable);具体的创建方式就是使用 Create Table xxx DDL 定义一个 Kafka 数据源输入表也可以是 Kafka 数据汇输出表。
xdm是不是又和 Hive 一样惊不惊喜意不意外。对比学习 1。
2.6 SQL 视图 VIEW
上文已经说了一个 VIEW 其实就是一段 SQL 逻辑的查询结果。
视图 VIEW 在 Table API 中的体现就是一个 Table 的 Java 对象其封装了一段查询逻辑。如下案例所示。
2.6.1 Table API 创建 VIEW
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;EnvironmentSettings settings EnvironmentSettings.newInstance().inStreamingMode() // 声明为流任务.build();TableEnvironment tEnv TableEnvironment.create(settings);// Table API 中的一个 Table 对象
Table projTable tEnv.from(X).select(...);// 将 projTable 创建为一个叫做 projectedTable 的 VIEW
tEnv.createTemporaryView(projectedTable, projTable);Table API 是使用了 TableEnvironment::createTemporaryView 接口将一个 Table 对象创建为一个 VIEW。
2.6.2 SQL API 创建 VIEW
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;EnvironmentSettings settings EnvironmentSettings.newInstance().inStreamingMode() // 声明为流任务.build();TableEnvironment tEnv TableEnvironment.create(settings);String sql CREATE TABLE source_table (\n user_id BIGINT,\n name STRING\n ) WITH (\n connector user_defined,\n format json,\n class.name flink.examples.sql._03.source_sink.table.user_defined.UserDefinedSource\n );\n \n CREATE TABLE sink_table (\n user_id BIGINT,\n name STRING\n ) WITH (\n connector print\n );\n CREATE VIEW query_view as\n // 创建 VIEW SELECT\n *\n FROM source_table\n ;\n INSERT INTO sink_table\n SELECT\n *\n FROM query_view;;Arrays.stream(sql.split(;)).forEach(tEnv::executeSql);SQL API 是直接通过一段 CREATE VIEW query_view as select * from source_table 来创建的 VIEW是纯 SQL 写法。
这种创建方式是不是贼熟悉和离线 Hive 一样。对比学习 1。 注意在 Table API 中的一个 Table 对象被后续的多个查询使用的场景下Table 对象不会真的产生一个中间表供下游多个查询去引用即多个查询不共享这个 Table 的结果小伙伴萌可以理解为是一种中间表的简化写法不会先产出一个中间表结果然后将这个结果在下游多个查询中复用后续的多个查询会将这个 Table 的逻辑执行多次。类似于 with tmp as (DML) 的语法 2.7 一个 SQL 查询案例
来看看一个 SQL 查询案例。
案例场景计算每一种商品sku_id 唯一标识的售出个数、总销售额、平均销售额、最低价、最高价。数据准备数据源为商品的销售流水sku_id商品price销售价格然后写入到 Kafka 的指定 topic 当中sku_id商品count_result售出个数、sum_result总销售额、avg_result平均销售额、min_result最低价、max_result最高价。
EnvironmentSettings settings EnvironmentSettings.newInstance().inStreamingMode() // 声明为流任务//.inBatchMode() // 声明为批任务.build();TableEnvironment tEnv TableEnvironment.create(settings);// 1. 创建一个数据源输入表这里的数据源是 flink 自带的一个随机 mock 数据的数据源。
String sourceSql CREATE TABLE source_table (\n sku_id STRING,\n price BIGINT\n ) WITH (\n connector datagen,\n rows-per-second 1,\n fields.sku_id.length 1,\n fields.price.min 1,\n fields.price.max 1000000\n );// 2. 创建一个数据汇输出表输出到 kafka 中
String sinkSql CREATE TABLE sink_table (\n sku_id STRING,\n count_result BIGINT,\n sum_result BIGINT,\n avg_result DOUBLE,\n min_result BIGINT,\n max_result BIGINT,\n PRIMARY KEY (sku_id) NOT ENFORCED\n ) WITH (\n connector upsert-kafka,\n topic tuzisir,\n properties.bootstrap.servers localhost:9092,\n key.format json,\n value.format json\n );// 3. 执行一段 group by 的聚合 SQL 查询
String selectWhereSql insert into sink_table\n select sku_id,\n count(*) as count_result,\n sum(price) as sum_result,\n avg(price) as avg_result,\n min(price) as min_result,\n max(price) as max_result\n from source_table\n group by sku_id;tEnv.executeSql(sourceSql);
tEnv.executeSql(sinkSql);
tEnv.executeSql(selectWhereSql);2.8 SQL 与 DataStream API 的转换
大家会比较好奇要写 SQL 就纯 SQL 呗要写 DataStream 就纯 DataStream 呗为啥还要把这两类接口做集成呢
博主举一个案例在 PDD 这种发补贴券的场景下希望可以在发的补贴券总金额超过 10000 10000 10000 元时及时报警出来来帮助控制预算防止发的太多。
对应的解决方案我们可以想到使用 SQL 计算补贴券发放的结果但是 SQL 的问题在于无法做到报警。所以我们可以将 SQL 的查询的结果即 Table 对象转为 DataStream然后就可以在 DataStream 后自定义报警逻辑的算子。
我们直接上 SQL 和 DataStream API 互相转化的案例
public static void main(String[] args) throws Exception {FlinkEnv flinkEnv FlinkEnvUtils.getStreamTableEnv(args);// 1. pdd 发补贴券流水数据String createTableSql CREATE TABLE source_table (\n id BIGINT,\n -- 补贴券的流水 id money BIGINT,\n -- 补贴券的金额 row_time AS cast(CURRENT_TIMESTAMP as timestamp_LTZ(3)),\n WATERMARK FOR row_time AS row_time - INTERVAL 5 SECOND\n ) WITH (\n connector datagen,\n rows-per-second 1,\n fields.id.min 1,\n fields.id.max 100000,\n fields.money.min 1,\n fields.money.max 100000\n )\n;// 2. 计算总计发放补贴券的金额String querySql SELECT UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, \n window_start, \n sum(money) as sum_money,\n -- 补贴券的发放总金额 count(distinct id) as count_distinct_id\n FROM TABLE(CUMULATE(\n TABLE source_table\n , DESCRIPTOR(row_time)\n , INTERVAL 5 SECOND\n , INTERVAL 1 DAY))\n GROUP BY window_start, \n window_end;flinkEnv.streamTEnv().executeSql(createTableSql);Table resultTable flinkEnv.streamTEnv().sqlQuery(querySql);// 3. 将金额结果转为 DataStream然后自定义超过 1w 的报警逻辑flinkEnv.streamTEnv().toDataStream(resultTable, Row.class).flatMap(new FlatMapFunctionRow, Object() {Overridepublic void flatMap(Row value, CollectorObject out) throws Exception {long l Long.parseLong(String.valueOf(value.getField(sum_money)));if (l 10000L) {log.info(报警超过 1w);}}});flinkEnv.env().execute();
}目前在 1.13 1.13 1.13 版本中Flink 对于 Table 和 DataStream 的转化是有一些限制的上面的案例可以看到Table 和 DataStream 之间的转换目前只有 StreamTableEnvironment::toDataStream、StreamTableEnvironment::fromDataStream 接口支持。
所以其实小伙伴萌可以理解为只有流任务才支持 Table 和 DataStream 之间的转换批任务是不支持的虽然可以使用流执行模式处理有界流 - 批数据也就是模拟按照批执行但效率较低这种骚操作不建议大家搞。
那什么时候才能支持批任务的 Table 和 DataStream 之间的转换呢 1.14 1.14 1.14 版本支持。 1.14 1.14 1.14 版本中流和批的都统一到了 StreamTableEnvironment 中因此就可以做 Table 和 DataStream 的互相转换了。
