NB
源码公众号后台回复1.13.2 最全 flink sql获取。
Apache Flink 提供了两种关系型 API 用于统一流和批处理,Table 和 SQL API。
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
DataSet<Row> result = tEnv
.from("Orders")
.groupBy($("a"))
.select($("a"), $("b").count().as("cnt"))
.toDataSet(counts, Row.class);
result.print();
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 去灵活的构建针对于异常数据的自定义报警机制。
在 xdm 大体了解了这两个 API 是干啥的之后,我们就可以直接来看看,怎么使用这两个 API 了。
根据小伙伴们使用的编程语言的不同(Java 或 Scala),需要将对应的依赖包添加到项目中。
<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>
<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 了。具体案例如下文所示。
在小伙伴萌看下文之前,先看一下 2.2 节整体的思路,跟着博主思路走,会更清晰:
// 创建一个 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 的能力有一个大概了解:
CREATE TABLE xxx
,INSERT INTO xxx SELECT xxx FROM xxx
。SELECT xxx FROM xxx
无论是对于 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 进行介绍了。
如果 xdm 想直接上手运行一段 Flink SQL 的代码。
可以直接在公众号后台回复1.13.2 最全 flink sql获取源代码。所有的源码都开源到 github 上面了。里面包含了非常多的案例。可以直接拿来在本地运行的!!!肥肠的方便。
TableEnvironment 是使用 SQL API 永远都离不开的一个接口。其是 SQL API 使用的入口(上下文),就像是你要使用 Java DataStream API 去写一个 Flink 任务需要使用到 StreamExecutionEnvironment 一样。
可以认为 TableEnvironment 在 SQL API 中的地位和 StreamExecutionEnvironment 在 DataStream 中的地位是一样的,都是包含了一个 Flink 任务运行时的所有上下文环境信息。大家这样对比学习会比较好理解。
TableEnvironment 包含的功能如下:
接下来介绍如何创建一个 TableEnvironment。案例为 Java。easy game。
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 版本中。
如果你是 inStreamingMode
,则最终创建出来的 TableEnvironment
实例为 StreamTableEnvironmentImpl
。
如果你是 inBatchMode
,则最终创建出来的 TableEnvironment
实例为 TableEnvironmentImpl
。
它两虽然都继承了 TableEnvironment
接口,但是 StreamTableEnvironmentImpl
支持的功能更多一些。大家可以直接去看看接口实验一下,这里就不进行详细介绍。
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
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 session 都生效。
-- 临时外部表
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 持久化!!!
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
;
由于目前在实时数据的场景中多以消息队列作为数据表。此处就以 Kafka 为例创建一个外部数据表。
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 数据源
DataStream<Row> r = env.addSource(new FlinkKafkaConsumer<Row>(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 对象来使用。
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 数据汇(输出)表)。
9
xdm,是不是又和 Hive 一样?惊不惊喜意不意外。对比学习 +1。
上文已经说了,一个 VIEW 其实就是一段 SQL 逻辑的查询结果。
视图 VIEW 在 Table API 中的体现就是:一个 Table 的 Java 对象,其封装了一段查询逻辑。如下案例所示:
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。
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 写法。
9
这种创建方式是不是贼熟悉,和离线 Hive 一样 +1~
注意: 在 Table API 中的一个 Table 对象被后续的多个查询使用的场景下:Table 对象不会真的产生一个中间表供下游多个查询去引用,即多个查询不共享这个 Table 的结果,小伙伴萌可以理解为是一种中间表的简化写法,不会先产出一个中间表结果,然后将这个结果在下游多个查询中复用,后续的多个查询会将这个 Table 的逻辑执行多次。类似于 with tmp as (DML) 的语法
首先,如果 xdm 想直接上手运行一段 Flink SQL 的代码。
可以直接在公众号后台回复1.13.2 最全 flink sql获取源代码。所有的源码都开源到 github 上面了。里面包含了非常多的案例。可以直接拿来在本地运行的!!!肥肠的方便。
来看看一个 SQL 查询案例。
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);
大家会比较好奇,要写 SQL 就纯 SQL 呗,要写 DataStream 就纯 DataStream 呗,为啥还要把这两类接口做集成呢?
博主举一个案例:在 pdd 这种发补贴券的场景下,希望可以在发的补贴券总金额超过 1w 元时,及时报警出来,来帮助控制预算,防止发的太多。
对应的解决方案,我们可以想到使用 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 FlatMapFunction<Row, Object>() {
@Override
public void flatMap(Row value, Collector<Object> out) throws Exception {
long l = Long.parseLong(String.valueOf(value.getField("sum_money")));
if (l > 10000L) {
log.info("报警,超过 1w");
}
}
});
flinkEnv.env().execute();
}
注意: 目前在 1.13 版本中,Flink 对于 Table 和 DataStream 的转化是有一些限制的:上面的案例可以看到,Table 和 DataStream 之间的转换目前只有
StreamTableEnvironment::toDataStream
、StreamTableEnvironment::fromDataStream
接口支持。 所以其实小伙伴萌可以理解为只有流任务才支持 Table 和 DataStream 之间的转换,批任务是不支持的(虽然可以使用流执行模式处理有界流 - 批数据,也就是模拟按照批执行,但效率较低,这种骚操作不建议大家搞)。 那什么时候才能支持批任务的 Table 和 DataStream 之间的转换呢?1.14 版本支持。1.14 版本中,流和批的都统一到了 StreamTableEnvironment 中,因此就可以做 Table 和 DataStream 的互相转换了。
在介绍完一些基本概念之后,我们来认识一下,Flink SQL 中的数据类型。
Flink SQL 内置了很多常见的数据类型,并且也为用户提供了自定义数据类型的能力。
总共包含 3 部分:
class
参数代表具体对应的 Java 类型,snapshot
代表类型在发生网络传输时的序列化器年-月-日
组成的 不带时区含义
的日期类型,取值范围 [0000-01-01, 9999-12-31]小时:分钟:秒[.小数秒]
组成的 不带时区含义
的的时间的数据类型,精度高达纳秒,取值范围 [00:00:00.000000000到23:59:59.9999999]。其中 p 代表小数秒的位数,取值范围 [0, 9],如果不指定 p,默认为 0。年-月-日 小时:分钟:秒[.小数秒]
组成的 不带时区含义
的时间类型,取值范围 [0000-01-01 00:00:00.000000000, 9999-12-31 23:59:59.999999999]。其中 p 代表小数秒的位数,取值范围 [0, 9],如果不指定 p,默认为 6。年-月-日 小时:分钟:秒[.小数秒] 时区
组成的 带时区含义
的时间类型,取值范围 [0000-01-01 00:00:00.000000000 +14:59, 9999-12-31 23:59:59.999999999 -14:59]。其中 p 代表小数秒的位数,取值范围 [0, 9],如果不指定 p,默认为 6。年-月-日 小时:分钟:秒[.小数秒] 时区
组成的 带时区含义
的时间类型,取值范围 [0000-01-01 00:00:00.000000000 +14:59, 9999-12-31 23:59:59.999999999 -14:59]。其中 p 代表小数秒的位数,取值范围 [0, 9],如果不指定 p,默认为 6。table.local-time-zone
参数来设置时区。CREATE TABLE sink_table (
result_interval_year TIMESTAMP(3),
result_interval_year_p TIMESTAMP(3),
result_interval_year_p_to_month TIMESTAMP(3),
result_interval_month TIMESTAMP(3),
result_interval_day TIMESTAMP(3),
result_interval_day_p1 TIMESTAMP(3),
result_interval_day_p1_to_hour TIMESTAMP(3),
result_interval_day_p1_to_minute TIMESTAMP(3),
result_interval_day_p1_to_second_p2 TIMESTAMP(3),
result_interval_hour TIMESTAMP(3),
result_interval_hour_to_minute TIMESTAMP(3),
result_interval_hour_to_second TIMESTAMP(3),
result_interval_minute TIMESTAMP(3),
result_interval_minute_to_second_p2 TIMESTAMP(3),
result_interval_second TIMESTAMP(3),
result_interval_second_p2 TIMESTAMP(3)
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT
-- Flink SQL 支持的所有 INTERVAL 子句如下,总体可以分为 `年-月`、`日-小时-秒` 两种
-- 1. 年-月。取值范围为 [-9999-11, +9999-11],其中 p 是指有效位数,取值范围 [1, 4],默认值为 2。比如如果值为 1000,但是 p = 2,则会直接报错。
-- INTERVAL YEAR
f1 + INTERVAL '10' YEAR as result_interval_year
-- INTERVAL YEAR(p)
, f1 + INTERVAL '100' YEAR(3) as result_interval_year_p
-- INTERVAL YEAR(p) TO MONTH
, f1 + INTERVAL '10-03' YEAR(3) TO MONTH as result_interval_year_p_to_month
-- INTERVAL MONTH
, f1 + INTERVAL '13' MONTH as result_interval_month
-- 2. 日-小时-秒。取值范围为 [-999999 23:59:59.999999999, +999999 23:59:59.999999999],其中 p1\p2 都是有效位数,p1 取值范围 [1, 6],默认值为 2;p2 取值范围 [0, 9],默认值为 6
-- INTERVAL DAY
, f1 + INTERVAL '10' DAY as result_interval_day
-- INTERVAL DAY(p1)
, f1 + INTERVAL '100' DAY(3) as result_interval_day_p1
-- INTERVAL DAY(p1) TO HOUR
, f1 + INTERVAL '10 03' DAY(3) TO HOUR as result_interval_day_p1_to_hour
-- INTERVAL DAY(p1) TO MINUTE
, f1 + INTERVAL '10 03:12' DAY(3) TO MINUTE as result_interval_day_p1_to_minute
-- INTERVAL DAY(p1) TO SECOND(p2)
, f1 + INTERVAL '10 00:00:00.004' DAY TO SECOND(3) as result_interval_day_p1_to_second_p2
-- INTERVAL HOUR
, f1 + INTERVAL '10' HOUR as result_interval_hour
-- INTERVAL HOUR TO MINUTE
, f1 + INTERVAL '10:03' HOUR TO MINUTE as result_interval_hour_to_minute
-- INTERVAL HOUR TO SECOND(p2)
, f1 + INTERVAL '00:00:00.004' HOUR TO SECOND(3) as result_interval_hour_to_second
-- INTERVAL MINUTE
, f1 + INTERVAL '10' MINUTE as result_interval_minute
-- INTERVAL MINUTE TO SECOND(p2)
, f1 + INTERVAL '05:05.006' MINUTE TO SECOND(3) as result_interval_minute_to_second_p2
-- INTERVAL SECOND
, f1 + INTERVAL '3' SECOND as result_interval_second
-- INTERVAL SECOND(p2)
, f1 + INTERVAL '300' SECOND(3) as result_interval_second_p2
FROM (SELECT TO_TIMESTAMP_LTZ(1640966476500, 3) as f1)
用户自定义类型就是运行用户使用 Java 等语言自定义一个数据类型出来。但是目前数据类型不支持使用 CREATE TABLE 的 DDL 进行定义,只支持作为函数的输入输出参数。如下案例:
public class User {
// 1. 基础类型,Flink 可以通过反射类型信息自动把数据类型获取到
// 关于 SQL 类型和 Java 类型之间的映射见:https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/types/#data-type-extraction
public int age;
public String name;
// 2. 复杂类型,用户可以通过 @DataTypeHint("DECIMAL(10, 2)") 注解标注此字段的数据类型
public @DataTypeHint("DECIMAL(10, 2)") BigDecimal totalBalance;
}
public class UserScalarFunction extends ScalarFunction {
// 1. 自定义数据类型作为输出参数
public User eval(long i) {
if (i > 0 && i <= 5) {
User u = new User();
u.age = (int) i;
u.name = "name1";
u.totalBalance = new BigDecimal(1.1d);
return u;
} else {
User u = new User();
u.age = (int) i;
u.name = "name2";
u.totalBalance = new BigDecimal(2.2d);
return u;
}
}
// 2. 自定义数据类型作为输入参数
public String eval(User i) {
if (i.age > 0 && i.age <= 5) {
User u = new User();
u.age = 1;
u.name = "name1";
u.totalBalance = new BigDecimal(1.1d);
return u.name;
} else {
User u = new User();
u.age = 2;
u.name = "name2";
u.totalBalance = new BigDecimal(2.2d);
return u.name;
}
}
}
-- 1. 创建 UDF
CREATE FUNCTION user_scalar_func AS 'flink.examples.sql._12_data_type._02_user_defined.UserScalarFunction';
-- 2. 创建数据源表
CREATE TABLE source_table (
user_id BIGINT NOT NULL COMMENT '用户 id'
) WITH (
'connector' = 'datagen',
'rows-per-second' = '1',
'fields.user_id.min' = '1',
'fields.user_id.max' = '10'
);
-- 3. 创建数据汇表
CREATE TABLE sink_table (
result_row_1 ROW<age INT, name STRING, totalBalance DECIMAL(10, 2)>,
result_row_2 STRING
) WITH (
'connector' = 'print'
);
-- 4. SQL 查询语句
INSERT INTO sink_table
select
-- 4.a. 用户自定义类型作为输出
user_scalar_func(user_id) as result_row_1,
-- 4.b. 用户自定义类型作为输出及输入
user_scalar_func(user_scalar_func(user_id)) as result_row_2
from source_table;
-- 5. 查询结果
+I[+I[9, name2, 2.20], name2]
+I[+I[1, name1, 1.10], name1]
+I[+I[5, name1, 1.10], name1]
在小伙伴萌看下文之前,先看一下 2.4 节整体的思路,跟着博主思路走,会更清晰:
SQL 动态输入表
技术来将 输入数据流
映射到 SQL 中的输入表
SQL 连续查询
技术来将 计算逻辑
映射到 SQL 中的运算语义
SQL 动态表 & 连续查询技术
两种技术方案来将 流式 SQL
实际应用到两个常见案例中SQL 连续查询
的两种类型:更新(Update)查询 & 追加(Append)查询SQL 动态输出表
技术来将 输出数据流
映射到 SQL 中的输出表
博主认为读完本节你应该掌握:
SQL 动态输入表
、SQL 动态输出表
SQL 连续查询
的两种类型分别对应的查询场景及 SQL 语义在流式 SQL 诞生之前,所有的基于 SQL 的数据查询都是基于批数据的,没有将 SQL 应用到流数据处理这一说法。
那么如果我们想将 SQL 应用到流处理中,必然要站在巨人的肩膀(批数据处理的流程)上面进行,那么具体的分析思路如下:
批处理
与 流处理
的异同之处:如果有相同的部分,那么可以直接复用;不同之处才是我们需要重点克服和关注的。博主下文就会根据上述三个步骤来一步一步介绍 动态表
诞生的背景以及这个概念是如何诞生的。
首先对比一下常见的 批处理
和 流处理
中 数据源(输入表)
、处理逻辑
、数据汇(结果表)
的异同点。
- | 输入表 | 处理逻辑 | 结果表 |
---|---|---|---|
批处理 | 静态表:输入数据有限、是有界集合 | 批式计算:每次执行查询能够访问到完整的输入数据,然后计算,输出完整的结果数据 | 静态表:数据有限 |
流处理 | 动态表:输入数据无限,数据实时增加,并且源源不断 | 流式计算:执行时不能够访问到完整的输入数据,每次计算的结果都是一个中间结果 | 动态表:数据无限 |
对比上述流批处理之后,我们得到了要将 SQL 应用于流式任务的三个要解决的核心点:
将上面 3 个点总结一下,也就引出了本节的 动态表
和 连续查询
两种技术方案:
动态表
:源源不断的输入、输出流数据映射到 动态表
连续查询
:实时处理输入数据,产出输出数据的实时处理技术动态表
。这里的动态其实是相比于批处理的静态(有界)来说的。
实时
进行变化的。一般都是一天、一小时的粒度新生成一个分区。来看一个具体的案例,下图显示了点击事件流
(左侧)如何转换为动态表
(右侧)。当数据源生成更多的点击事件记录时,映射出来的动态表也会不断增长,这就是动态表的概念:
Dynamic Table
连续查询
。
部分高级关系数据库系统提供了一个称为物化视图(Materialized Views) 的特性。
物化视图其实就是一条 SQL 查询,就像常规的虚拟视图 VIEW 一样。但与虚拟视图不同的是,物化视图会缓存查询的结果,因此在请求访问视图时不需要对查询进行重新计算,可以直接获取物化视图的结果,小伙伴萌可以认为物化视图其实就是把结果缓存了下来。
举个例子:批处理中,如果以 Hive 天级别的物化视图来说,其实就是每天等数据源 ready 之后,调度物化视图的 SQL 执行然后产生新的结果提供服务。那么就可以认为一条表示了输入、处理、输出的 SQL 就是一个构建物化视图的过程。
映射到我们的流任务中,输入、处理逻辑、输出这一套流程也是一个物化视图的概念。相比批处理来说,流处理中,我们的数据源表的数据是源源不断的。那么从输入、处理、输出的整个物化视图的维护流程也必须是实时的。
因此我们就需要引入一种实时视图维护(Eager View Maintenance)
的技术去做到:一旦更新了物化视图的数据源表就立即更新视图的结果,从而保证输出的结果也是最新的。
这种 实时视图维护(Eager View Maintenance)
的技术就叫做 连续查询
。
注意:
总结前两节,动态表
& 连续查询
两项技术在一条流 SQL 中的执行流程总共包含了三个步骤,如下图及总结所示:
Query
我们实际介绍一个案例来看看其运行方式,以上文介绍到的点击事件流为例,点击事件流数据的字段如下:
[
user: VARCHAR, // 用户名
cTime: TIMESTAMP, // 访问 URL 的时间
url: VARCHAR // 用户访问的 URL
]
Dynamic Table
下面介绍两个查询的案例:
第一个查询:一个简单的 GROUP-BY COUNT 聚合查询,写过 SQL 的都不会陌生吧,这种应该都是最基础,最常用的对数据按照类别分组的方法。
如下图所示 group by 聚合的常用案例。
time
那么本案例中呢,是基于 clicks 表中 user 字段对 clicks 表(点击事件流)进行分组,来统计每一个 user 的访问的 URL 的数量。下面的图展示了当 clicks 输入表来了新数据(即表更新时),连续查询(Continuous Query) 的计算逻辑。
group agg
当查询开始,clicks 表(左侧)是空的。
插入(insert)结果表
。插入(insert)到结果表
。更新(update)结果表
,[Mary, 1] 更新成 [Mary, 2]。插入(insert)结果表
中。注意上述特殊标记出来的字体,可以看到连续查询对于结果的数据输出方式有两种:
大家对于 插入(insert)结果表
这件事都比较好理解,因为离线数据都只有插入这个概念。
但是 更新(update)结果表
就是离线处理中没有概念了。这就是连续查询中中比较重要一个概念。后文会介绍。
接下来介绍第二条查询语句。
第二条查询与第一条类似
,但是 group by 中除了 user 字段之外,还 group by 了 tumble,其代表开了个滚动窗口(后面会详细说明滚动窗口的作用),然后计算 url 数量。
group by user,是按照类别(横向)给数据分组,group by tumble 滚动窗口是按时间粒度(纵向)给数据进行分组。如下图所示。
time
图形化一解释就很好理解了,两种都是对数据进行分组,一个是按照 类别
分组,另一种是按照 时间
分组。
与前面一样,左边显示了输入表 clicks。查询每小时持续计算结果并更新结果表。clicks 表有三列,user,cTime,url。其中 cTime 代表数据的时间戳,用于给数据按照时间粒度分组。
tumble window
我们的滚动窗口的步长为 1 小时,即时间粒度上面的分组为 1 小时。其中时间戳在 12:00:00 - 12:59:59 之间有四条数据。13:00:00 - 13:59:59 有三条数据。14:00:00 - 14:59:59 之间有四条数据。
插入(insert)结果表
。插入(insert)结果表
。插入(insert)结果表
。而这个查询只有 插入(insert)结果表
这个行为。
虽然前一节的两个查询看起来非常相似(都计算分组进行计数聚合),但它们在一个重要方面不同:
上面是 Flink SQL 连续查询处理机制上面的两类查询方式。我们可以发现连续查询的处理机制不一样,产出到结果表中的结果数据也是不一样的。针对上面两种结果表的更新方式,Flink SQL 提出了 changelog 表的概念来进行兼容。
changelog 表这个概念其实就和 MySQL binlog 是一样的。会包含 INSERT
、UPDATE
、DELETE
三种数据,通过这三种数据的处理来描述实时处理技术对于动态表的变更:
可以看到我们的标题都是随着一个 SQL 的生命周期的。从 输入流映射为 SQL 动态输入表
、实时处理底层技术 - SQL 连续查询
到本小节的 SQL 动态输出表转化为输出数据
。都是有逻辑关系的。
我们上面介绍到了 连续查询(Continuous Query) 的输出结果表是一个 changelog。其可以像普通数据库表一样通过 INSERT、UPDATE 和 DELETE 来不断修改。
它可能是一个只有一行、不断更新 changelog 表,也可能是一个 insert-only 的 changelog 表,没有 UPDATE 和 DELETE 修改,或者介于两者之间的其他表。
在将动态表转换为流或将其写入外部系统时,需要对这些不同状态的数据进行编码。Flink 的 Table API 和 SQL API 支持三种方式来编码一个动态表的变化:
INSERT
操作的数据。INSERT
操作编码为 add message、将 DELETE
操作编码为 retract message、将 UPDATE
操作编码为更新先前行的 retract message 和更新(新)行的 add message,从而将动态表转换为 retract 流。-
,+
两种,分别 -
代表撤回旧数据,+
代表输出最新的数据。这两种数据最终都会写入到输出的数据引擎中。-
,+
两种数据,防止数据计算重复或者错误。retract
INSERT
和 UPDATE
操作编码为 upsert message,将 DELETE
操作编码为 delete message。-
回撤数据。upsert
小伙伴萌会问到,关系代数是啥东西?
其实关系代数就是对于数据集(即表)的一系列的 操作
(即查询语句)。常见关系代数有:
Relational Algebra
⭐ 那么 SQL 和关系代数是啥关系呢?
SQL 就是能够表示关系代数一种面向用户的接口:即用户能使用 SQL 表达关系代数的处理逻辑,也就是我们可以用 SQL 去在表(数据集)上执行我们的业务逻辑操作(关系代数操作)。
在小伙伴萌看下文之前,先看一下 2.5 节整体的思路,跟着博主思路走:
处理时间
、事件时间
、摄入时间
三种时间语义。事件时间(SQL 常用)
> 处理时间(SQL 几乎不用,DataStream 少用)
> 摄入时间(不用)
time
事件时间
。目前生产环境中用的最多
。在生产环境中用的次多
。摄入时间用的最少,可以说基本不使用
。小伙伴萌要注意到:
讲到这里,xdm 会问,博主上面写的 3 种时间属性到底对我们的任务有啥影响呢?3 种时间属性的应用场景是啥?
先说结论,在 Flink 中时间的作用:
主要体现在包含时间窗口的计算中
:用于标识任务的时间进度,来判断是否需要触发窗口的计算。比如常用的滚动窗口
、滑动窗口
等都需要时间推动触发。这些窗口的应用场景后续会详细介绍。次要体现在自定义时间语义的计算中
:举个例子,比如用户可以自定义每隔 10s 的本地时间,或者消费到的数据的时间戳每增大 10s,就把计算结果输出一次,时间在此类应用中也是一种标识任务进度的作用。博主以 滚动窗口
的聚合任务为例来介绍一下事件时间和处理时间的对比区别。
tumble window
上面这个案例的窗口大小是 1 小时,需求方需要按照用户点击时间戳 cTime
划分数据(划分滚动窗口),然后计算出 count 聚合结果(这样计算能反映出事件的真实发生时间),那么就需要把 cTime
设置为窗口的划分时间戳,即代码中 tumble(cTime, interval '1' hour)
。
上面这种就叫做事件时间。即用数据中自带的时间戳进行窗口的划分(点击操作真实的发生时间)。
后续 Flink SQL 任务在运行的过程中也会实际按照 cTime
的当前时间作为一小时窗口结束触发条件并计算一个小时窗口内的数据。
还是上面那个案例,但是这次需求方不需要按照数据上的时间戳划分数据(划分滚动窗口),只需要数据来了之后, 在 Flink 机器上的时间作为一小时窗口结束的书法条件并计算。
那么这种触发机制就是处理时间。
如果要满足 Flink SQL 时间窗口类的聚合操作,SQL 或 Table API 中的 数据源表
就需要提供时间属性(相当于我们把这个时间属性在 数据源表
上面进行声明),以及支持时间相关的操作。
那么来看看 Flink SQL 为我们提供的两种指定时间戳的方式:
CREATE TABLE DDL
创建表的时候指定可以在 DataStream 中指定
,在后续的 DataStream 转的 Table 中使用一旦时间属性定义好,它就可以像普通列一样使用,也可以在时间相关的操作中使用。
来看看 Flink 中如何指定事件时间。
CREATE TABLE DDL
指定时间戳的方式。CREATE TABLE user_actions (
user_name STRING,
data STRING,
user_action_time TIMESTAMP(3),
-- 使用下面这句来将 user_action_time 声明为事件时间,并且声明 watermark 的生成规则,即 user_action_time 减 5 秒
-- 事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型
WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
...
);
SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
-- 然后就可以在窗口算子中使用 user_action_time
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);
从上面这条语句可以看到,如果想使用事件时间,那么我们的时间戳类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型。很多小伙伴会想到,我们的时间戳一般不都是秒或者是毫秒(BIGINT 类型)嘛,那这种情况怎么办?
解决方案必须要有啊。如下。
CREATE TABLE user_actions (
user_name STRING,
data STRING,
-- 1. 这个 ts 就是常见的毫秒级别时间戳
ts BIGINT,
-- 2. 将毫秒时间戳转换成 TIMESTAMP_LTZ 类型
time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
-- 3. 使用下面这句来将 user_action_time 声明为事件时间,并且声明 watermark 的生成规则,即 user_action_time 减 5 秒
-- 事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型
WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND
) WITH (
...
);
SELECT TUMBLE_START(time_ltz, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(time_ltz, INTERVAL '10' MINUTE);
DataStream
中指定事件时间。之前介绍了 Table
和 DataStream
可以互转,那么 Flink 也提供了一个能力,就是在 Table 转为 DataStream 时,指定时间戳字段。如下案例:
public class DataStreamSourceEventTimeTest {
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);
// 1. 分配 watermark
DataStream<Row> r = env.addSource(new UserDefinedSource())
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Row>(Time.minutes(0L)) {
@Override
public long extractTimestamp(Row element) {
return (long) element.getField("f2");
}
});
// 2. 使用 f2.rowtime 的方式将 f2 字段指为事件时间时间戳
Table sourceTable = tEnv.fromDataStream(r, "f0, f1, f2.rowtime");
tEnv.createTemporaryView("source_table", sourceTable);
// 3. 在 tumble window 中使用 f2
String tumbleWindowSql =
"SELECT TUMBLE_START(f2, INTERVAL '5' SECOND), COUNT(DISTINCT f0)\n"
+ "FROM source_table\n"
+ "GROUP BY TUMBLE(f2, INTERVAL '5' SECOND)"
;
Table resultTable = tEnv.sqlQuery(tumbleWindowSql);
tEnv.toDataStream(resultTable, Row.class).print();
env.execute();
}
private static class UserDefinedSource implements SourceFunction<Row>, ResultTypeQueryable<Row> {
private volatile boolean isCancel;
@Override
public void run(SourceContext<Row> sourceContext) throws Exception {
int i = 0;
while (!this.isCancel) {
sourceContext.collect(Row.of("a" + i, "b", System.currentTimeMillis()));
Thread.sleep(10L);
i++;
}
}
@Override
public void cancel() {
this.isCancel = true;
}
@Override
public TypeInformation<Row> getProducedType() {
return new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),
TypeInformation.of(Long.class));
}
}
}
来看看 Flink SQL 中如何指定处理时间。
CREATE TABLE DDL
指定时间戳的方式。CREATE TABLE user_actions (
user_name STRING,
data STRING,
-- 使用下面这句来将 user_action_time 声明为处理时间
user_action_time AS PROCTIME()
) WITH (
...
);
SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
-- 然后就可以在窗口算子中使用 user_action_time
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);
DataStream
中指定处理时间。public class DataStreamSourceProcessingTimeTest {
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);
// 1. 分配 watermark
DataStream<Row> r = env.addSource(new UserDefinedSource());
// 2. 使用 proctime.proctime 的方式将 f2 字段指为处理时间时间戳
Table sourceTable = tEnv.fromDataStream(r, "f0, f1, f2, proctime.proctime");
tEnv.createTemporaryView("source_table", sourceTable);
// 3. 在 tumble window 中使用 f2
String tumbleWindowSql =
"SELECT TUMBLE_START(proctime, INTERVAL '5' SECOND), COUNT(DISTINCT f0)\n"
+ "FROM source_table\n"
+ "GROUP BY TUMBLE(proctime, INTERVAL '5' SECOND)"
;
Table resultTable = tEnv.sqlQuery(tumbleWindowSql);
tEnv.toDataStream(resultTable, Row.class).print();
env.execute();
}
private static class UserDefinedSource implements SourceFunction<Row>, ResultTypeQueryable<Row> {
private volatile boolean isCancel;
@Override
public void run(SourceContext<Row> sourceContext) throws Exception {
int i = 0;
while (!this.isCancel) {
sourceContext.collect(Row.of("a" + i, "b", System.currentTimeMillis()));
Thread.sleep(10L);
i++;
}
}
@Override
public void cancel() {
this.isCancel = true;
}
@Override
public TypeInformation<Row> getProducedType() {
return new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),
TypeInformation.of(Long.class));
}
}
}
首先说一下这个问题的背景:
大家想一下离线 Hive 环境中,有遇到过时区时区相关的问题吗?
至少博主目前没有碰到过,因为这个问题在底层的数据集成系统都已经给解决了,小伙伴萌拿到手的 ODS 层表都是已经按照所在地区的时区给格式化好的了。
举个例子:小伙伴萌看到日期分区为 2022-01-01 的 Hive 表时,可以默认认为该分区中的数据就对应到你所在地区的时区的 2022-01-01 日的数据。
但是 Flink 中时区问题要特别引起关注,不加小心就会误用。
而本节 SQL 时区旨在帮助大家了解到以下两个场景的问题:
PROCTIME()
指定处理时间列时,返回值类型为 TIMESTAMP(3) 类型,而 TIMESTAMP(3) 是不带任何时区信息的,默认为 UTC 时间(0 时区)。StreamTableEnvironment::createTemporaryView
将 DataStream 转为 Table 时,注册处理时间(proctime.proctime
)、事件时间列(rowtime.rowtime
)时,两列时间类型也为 TIMESTAMP(3) 类型,不带时区信息。而以上两个场景就会导致:
因此充分了解本节的知识内容可以很好的帮你避免时区问题错误。
年, 月, 日, 小时, 分钟, 秒 和 小数秒
的字符串来指定。举例:1970-01-01 00:00:04.001。table.local-time-zone
设置的table.local-time-zone
时区参数之后,就能将一个不带有时区信息的时间戳转换为带有时区信息的字符串了。举例:table.local-time-zone
为 Asia/Shanghai
时,4001 时间戳转化为字符串的效果是 1970-01-01 08:00:04.001
。如果你还对时区问题有疑惑,可以参考博主写的一篇时区相关的文章。
以下 SQL 中的时间函数都会受到时区参数的影响,从而做到最后显示给用户的时间、窗口的划分都按照用户设置时区之内的时间。
在 Flink SQL client 中执行结果如下:
Flink SQL> SET sql-client.execution.result-mode=tableau;
Flink SQL> CREATE VIEW MyView1 AS SELECT LOCALTIME, LOCALTIMESTAMP, CURRENT_DATE, CURRENT_TIME, CURRENT_TIMESTAMP, CURRENT_ROW_TIMESTAMP(), NOW(), PROCTIME();
Flink SQL> DESC MyView1;
+------------------------+-----------------------------+-------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+------------------------+-----------------------------+-------+-----+--------+-----------+
| LOCALTIME | TIME(0) | false | | | |
| LOCALTIMESTAMP | TIMESTAMP(3) | false | | | |
| CURRENT_DATE | DATE | false | | | |
| CURRENT_TIME | TIME(0) | false | | | |
| CURRENT_TIMESTAMP | TIMESTAMP_LTZ(3) | false | | | |
|CURRENT_ROW_TIMESTAMP() | TIMESTAMP_LTZ(3) | false | | | |
| NOW() | TIMESTAMP_LTZ(3) | false | | | |
| PROCTIME() | TIMESTAMP_LTZ(3) *PROCTIME* | false | | | |
+------------------------+-----------------------------+-------+-----+--------+-----------+
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT * FROM MyView1;
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| LOCALTIME | LOCALTIMESTAMP | CURRENT_DATE | CURRENT_TIME | CURRENT_TIMESTAMP | CURRENT_ROW_TIMESTAMP() | NOW() | PROCTIME() |
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| 15:18:36 | 2021-04-15 15:18:36.384 | 2021-04-15 | 15:18:36 | 2021-04-15 15:18:36.384 | 2021-04-15 15:18:36.384 | 2021-04-15 15:18:36.384 | 2021-04-15 15:18:36.384 |
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT * FROM MyView1;
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| LOCALTIME | LOCALTIMESTAMP | CURRENT_DATE | CURRENT_TIME | CURRENT_TIMESTAMP | CURRENT_ROW_TIMESTAMP() | NOW() | PROCTIME() |
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| 23:18:36 | 2021-04-15 23:18:36.384 | 2021-04-15 | 23:18:36 | 2021-04-15 23:18:36.384 | 2021-04-15 23:18:36.384 | 2021-04-15 23:18:36.384 | 2021-04-15 23:18:36.384 |
+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
Flink SQL> CREATE VIEW MyView2 AS SELECT TO_TIMESTAMP_LTZ(4001, 3) AS ltz, TIMESTAMP '1970-01-01 00:00:01.001' AS ntz;
Flink SQL> DESC MyView2;
+------+------------------+-------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+------+------------------+-------+-----+--------+-----------+
| ltz | TIMESTAMP_LTZ(3) | true | | | |
| ntz | TIMESTAMP(3) | false | | | |
+------+------------------+-------+-----+--------+-----------+
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT * FROM MyView2;
+-------------------------+-------------------------+
| ltz | ntz |
+-------------------------+-------------------------+
| 1970-01-01 00:00:04.001 | 1970-01-01 00:00:01.001 |
+-------------------------+-------------------------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT * FROM MyView2;
+-------------------------+-------------------------+
| ltz | ntz |
+-------------------------+-------------------------+
| 1970-01-01 08:00:04.001 | 1970-01-01 00:00:01.001 |
+-------------------------+-------------------------+
Flink SQL> CREATE VIEW MyView3 AS SELECT ltz, CAST(ltz AS TIMESTAMP(3)), CAST(ltz AS STRING), ntz, CAST(ntz AS TIMESTAMP_LTZ(3)) FROM MyView2;
Flink SQL> DESC MyView3;
+-------------------------------+------------------+-------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+-------------------------------+------------------+-------+-----+--------+-----------+
| ltz | TIMESTAMP_LTZ(3) | true | | | |
| CAST(ltz AS TIMESTAMP(3)) | TIMESTAMP(3) | true | | | |
| CAST(ltz AS STRING) | STRING | true | | | |
| ntz | TIMESTAMP(3) | false | | | |
| CAST(ntz AS TIMESTAMP_LTZ(3)) | TIMESTAMP_LTZ(3) | false | | | |
+-------------------------------+------------------+-------+-----+--------+-----------+
Flink SQL> SELECT * FROM MyView3;
+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+
| ltz | CAST(ltz AS TIMESTAMP(3)) | CAST(ltz AS STRING) | ntz | CAST(ntz AS TIMESTAMP_LTZ(3)) |
+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+
| 1970-01-01 08:00:04.001 | 1970-01-01 08:00:04.001 | 1970-01-01 08:00:04.001 | 1970-01-01 00:00:01.001 | 1970-01-01 00:00:01.001 |
+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+
这里分两类,分别是 TIMESTAMP(不带时区信息的时间)、TIMESTAMP_LTZ(带时区信息的时间) 的事件时间 Flink SQL 任务
Flink SQL> CREATE TABLE MyTable2 (
item STRING,
price DOUBLE,
ts TIMESTAMP(3), -- TIMESTAMP 类型的时间戳
WATERMARK FOR ts AS ts - INTERVAL '10' SECOND
) WITH (
'connector' = 'socket',
'hostname' = '127.0.0.1',
'port' = '9999',
'format' = 'csv'
);
Flink SQL> CREATE VIEW MyView4 AS
SELECT
TUMBLE_START(ts, INTERVAL '10' MINUTES) AS window_start,
TUMBLE_END(ts, INTERVAL '10' MINUTES) AS window_end,
TUMBLE_ROWTIME(ts, INTERVAL '10' MINUTES) as window_rowtime,
item,
MAX(price) as max_price
FROM MyTable2
GROUP BY TUMBLE(ts, INTERVAL '10' MINUTES), item;
Flink SQL> DESC MyView4;
+----------------+------------------------+------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+----------------+------------------------+------+-----+--------+-----------+
| window_start | TIMESTAMP(3) | true | | | |
| window_end | TIMESTAMP(3) | true | | | |
| window_rowtime | TIMESTAMP(3) *ROWTIME* | true | | | |
| item | STRING | true | | | |
| max_price | DOUBLE | true | | | |
+----------------+------------------------+------+-----+--------+-----------+
将数据写入到 MyTable2 中:
> nc -lk 9999
A,1.1,2021-04-15 14:01:00
B,1.2,2021-04-15 14:02:00
A,1.8,2021-04-15 14:03:00
B,2.5,2021-04-15 14:04:00
C,3.8,2021-04-15 14:05:00
C,3.8,2021-04-15 14:11:00
最终结果如下:
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT * FROM MyView4;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_rowtime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | A | 1.8 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | B | 2.5 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT * FROM MyView4;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_rowtime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | A | 1.8 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | B | 2.5 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
通过上述结果可见,使用 TIMESTAMP(不带时区信息的时间) 进开窗,在 UTC 时区下的计算结果与在 Asia/Shanghai 时区下计算的窗口开始时间,窗口结束时间和窗口的时间是相同的。
Flink SQL> CREATE TABLE MyTable3 (
item STRING,
price DOUBLE,
ts BIGINT, -- long 类型的时间戳
ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3), -- 转为 TIMESTAMP_LTZ 类型的时间戳
WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '10' SECOND
) WITH (
'connector' = 'socket',
'hostname' = '127.0.0.1',
'port' = '9999',
'format' = 'csv'
);
Flink SQL> CREATE VIEW MyView5 AS
SELECT
TUMBLE_START(ts_ltz, INTERVAL '10' MINUTES) AS window_start,
TUMBLE_END(ts_ltz, INTERVAL '10' MINUTES) AS window_end,
TUMBLE_ROWTIME(ts_ltz, INTERVAL '10' MINUTES) as window_rowtime,
item,
MAX(price) as max_price
FROM MyTable3
GROUP BY TUMBLE(ts_ltz, INTERVAL '10' MINUTES), item;
Flink SQL> DESC MyView5;
+----------------+----------------------------+-------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+----------------+----------------------------+-------+-----+--------+-----------+
| window_start | TIMESTAMP(3) | false | | | |
| window_end | TIMESTAMP(3) | false | | | |
| window_rowtime | TIMESTAMP_LTZ(3) *ROWTIME* | true | | | |
| item | STRING | true | | | |
| max_price | DOUBLE | true | | | |
+----------------+----------------------------+-------+-----+--------+-----------+
将数据写入 MyTable3:
A,1.1,1618495260000 # 对应到 UTC 时区的时间为 2021-04-15 14:01:00
B,1.2,1618495320000 # 对应到 UTC 时区的时间为 2021-04-15 14:02:00
A,1.8,1618495380000 # 对应到 UTC 时区的时间为 2021-04-15 14:03:00
B,2.5,1618495440000 # 对应到 UTC 时区的时间为 2021-04-15 14:04:00
C,3.8,1618495500000 # 对应到 UTC 时区的时间为 2021-04-15 14:05:00
C,3.8,1618495860000 # 对应到 UTC 时区的时间为 2021-04-15 14:11:00
最终结果如下:
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT * FROM MyView5;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_rowtime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | A | 1.8 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | B | 2.5 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:09:59.999 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT * FROM MyView5;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_rowtime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:09:59.999 | A | 1.8 |
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:09:59.999 | B | 2.5 |
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:09:59.999 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
通过上述结果可见,使用 TIMESTAMP_LTZ(带时区信息的时间) 进开窗,在 UTC 时区下的计算结果与在 Asia/Shanghai 时区下计算的窗口开始时间,窗口结束时间和窗口的时间是不同的,都是按照时区进行格式化的。
Flink SQL 定义处理时间属性列是通过 PROCTIME()
函数来指定的,其返回值类型是 TIMESTAMP_LTZ。
注意: 在 Flink 1.13 之前,
PROCTIME()
函数返回类型是 TIMESTAMP,返回值是 UTC 时区的时间戳,例如,上海时间显示为 2021-03-01 12:00:00 时,PROCTIME() 返回值显示 2021-03-01 04:00:00,我们进行使用是错误的。Flink 1.13 修复了这个问题,使用 TIMESTAMP_LTZ 作为 PROCTIME() 的返回类型,这样 Flink 就会自动获取当前时区信息,然后进行处理,不需要用户再进行时区的格式化处理了。
如下案例:
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT PROCTIME();
+-------------------------+
| PROCTIME() |
+-------------------------+
| 2021-04-15 14:48:31.387 |
+-------------------------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT PROCTIME();
+-------------------------+
| PROCTIME() |
+-------------------------+
| 2021-04-15 22:48:31.387 |
+-------------------------+
Flink SQL> CREATE TABLE MyTable1 (
item STRING,
price DOUBLE,
proctime as PROCTIME()
) WITH (
'connector' = 'socket',
'hostname' = '127.0.0.1',
'port' = '9999',
'format' = 'csv'
);
Flink SQL> CREATE VIEW MyView3 AS
SELECT
TUMBLE_START(proctime, INTERVAL '10' MINUTES) AS window_start,
TUMBLE_END(proctime, INTERVAL '10' MINUTES) AS window_end,
TUMBLE_PROCTIME(proctime, INTERVAL '10' MINUTES) as window_proctime,
item,
MAX(price) as max_price
FROM MyTable1
GROUP BY TUMBLE(proctime, INTERVAL '10' MINUTES), item;
Flink SQL> DESC MyView3;
+-----------------+-----------------------------+-------+-----+--------+-----------+
| name | type | null | key | extras | watermark |
+-----------------+-----------------------------+-------+-----+--------+-----------+
| window_start | TIMESTAMP(3) | false | | | |
| window_end | TIMESTAMP(3) | false | | | |
| window_proctime | TIMESTAMP_LTZ(3) *PROCTIME* | false | | | |
| item | STRING | true | | | |
| max_price | DOUBLE | true | | | |
+-----------------+-----------------------------+-------+-----+--------+-----------+
将数据写入到 MyTable1 中:
> nc -lk 9999
A,1.1
B,1.2
A,1.8
B,2.5
C,3.8
其输出结果如下:
Flink SQL> SET table.local-time-zone=UTC;
Flink SQL> SELECT * FROM MyView3;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_procime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:10:00.005 | A | 1.8 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:10:00.007 | B | 2.5 |
| 2021-04-15 14:00:00.000 | 2021-04-15 14:10:00.000 | 2021-04-15 14:10:00.007 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
Flink SQL> SET table.local-time-zone=Asia/Shanghai;
Flink SQL> SELECT * FROM MyView3;
+-------------------------+-------------------------+-------------------------+------+-----------+
| window_start | window_end | window_procime | item | max_price |
+-------------------------+-------------------------+-------------------------+------+-----------+
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:10:00.005 | A | 1.8 |
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:10:00.007 | B | 2.5 |
| 2021-04-15 22:00:00.000 | 2021-04-15 22:10:00.000 | 2021-04-15 22:10:00.007 | C | 3.8 |
+-------------------------+-------------------------+-------------------------+------+-----------+
通过上述结果可见,使用处理时间进行开窗,在 UTC 时区下的计算结果与在 Asia/Shanghai 时区下计算的窗口开始时间,窗口结束时间和窗口的时间是不同的,都是按照时区进行格式化的。
以下函数:
在 Streaming 模式下这些函数是每条记录都会计算一次,但在 Batch 模式下,只会在 query 开始时计算一次,所有记录都使用相同的时间结果。
以下时间函数无论是在 Streaming 模式还是 Batch 模式下,都会为每条记录计算一次结果: