陕西网站开发哪家好,谷歌浏览器怎么下载,个人征信查询,做论坛网站需要什么备案《Flink SQL 基础概念》系列#xff0c;共包含以下 5 篇文章#xff1a;
Flink SQL 基础概念#xff08;一#xff09;#xff1a;SQL Table 运行环境、基本概念及常用 APIFlink SQL 基础概念#xff08;二#xff09;#xff1a;数据类型Flink SQL 基础概念共包含以下 5 篇文章
Flink SQL 基础概念一SQL Table 运行环境、基本概念及常用 APIFlink SQL 基础概念二数据类型Flink SQL 基础概念三SQL 动态表 连续查询Flink SQL 基础概念四SQL 的时间属性Flink SQL 基础概念五SQL 时区问题 如果您觉得这篇文章有用 ✔️ 的话请给博主一个一键三连 吧 点赞 、关注 、收藏 您的支持 将激励 博主输出更多优质内容 Flink SQL 基础概念四SQL 的时间属性 1.Flink 三种时间属性简介2.Flink 三种时间属性的应用场景2.1 事件时间案例2.2 处理时间案例2.3 摄入时间案例 3.SQL 指定时间属性的两种方式4.SQL 事件时间案例5.SQL 处理时间案例 与离线处理中常见的时间分区字段一样在实时处理中时间属性也是一个核心概念。Flink 支持 处理时间、事件时间、摄入时间 三种时间语义。
三种时间在生产环境的使用频次 事件时间SQL 常用 处理时间SQL 几乎不用DataStream 少用 摄入时间不用。
1.Flink 三种时间属性简介
事件时间指的是数据本身携带的时间这个时间是在事件产生时的时间而且在 Flink SQL 触发计算时也使用数据本身携带的时间。这就叫做事件时间。目前生产环境中用的最多。处理时间指的是具体算子计算数据执行时的机器时间例如在算子中 Java 取 System.currentTimeMillis()在生产环境中用的次多。摄入时间指的是数据从数据源进入 Flink 的时间。摄入时间用的最少可以说基本不使用。
小伙伴们要注意到
上述的三种时间概念不是由于有了数据而诞生的而是有了 Flink 之后根据实际的应用场景而诞生的。以事件时间举个例子如果只是数据携带了时间Flink 也消费了这个数据但是在 Flink 中没有使用数据的这个时间作为计算的触发条件也不能把这个 Flink 任务叫做事件时间的任务。其次要认识到一般一个 Flink 任务只会有一个时间属性所以时间属性通常认为是一个任务粒度的。举例我们可以说 A 任务是事件时间语义的任务B 任务是处理时间语义的任务。当然了一个任务也可以存在多个时间属性。
2.Flink 三种时间属性的应用场景
讲到这里有人会问博主上面写的 3 种时间属性到底对我们的任务有啥影响呢3 种时间属性的应用场景是啥
先说结论在 Flink 中时间的作用
主要体现在包含时间窗口的计算中用于标识任务的时间进度来判断是否需要触发窗口的计算。比如常用的滚动窗口、滑动窗口等都需要时间推动触发。这些窗口的应用场景后续会详细介绍。次要体现在自定义时间语义的计算中举个例子比如用户可以自定义每隔 10s 的本地时间或者消费到的数据的时间戳每增大 10s就把计算结果输出一次时间在此类应用中也是一种标识任务进度的作用。
博主以 滚动窗口 的聚合任务为例来介绍一下事件时间和处理时间的对比区别。
2.1 事件时间案例
还是以之前的 clicks 表拿来举例。 上面这个案例的窗口大小是 1 小时需求方需要按照用户点击时间戳 cTime 划分数据划分滚动窗口然后计算出 Count 聚合结果这样计算能反映出事件的真实发生时间那么就需要把 cTime 设置为窗口的划分时间戳即代码中 tumble(cTime, interval 1 hour)。
上面这种就叫做事件时间。即用数据中自带的时间戳进行窗口的划分点击操作真实的发生时间。
后续 Flink SQL 任务在运行的过程中也会实际按照 cTime 的当前时间作为一小时窗口结束触发条件并计算一个小时窗口内的数据。
2.2 处理时间案例
还是以之前的 clicks 表拿来举例。
还是上面那个案例但是这次需求方不需要按照数据上的时间戳划分数据划分滚动窗口只需要数据来了之后 在 Flink 机器上的时间作为一小时窗口结束的触发条件并计算。
那么这种触发机制就是处理时间。
2.3 摄入时间案例
在 Flink 从外部数据源读取到数据时给这条数据带上的当前数据源算子的本地时间戳。下游可以用这个时间戳进行窗口聚合不过这种几乎不使用。
3.SQL 指定时间属性的两种方式
如果要满足 Flink SQL 时间窗口类的聚合操作SQL 或 Table API 中的 数据源表 就需要提供时间属性相当于我们把这个时间属性在 数据源表 上面进行声明以及支持时间相关的操作。
那么来看看 Flink SQL 为我们提供的两种指定时间戳的方式
CREATE TABLE DDL 创建表的时候指定可以在 DataStream 中指定在后续的 DataStream 转的 Table 中使用
一旦时间属性定义好它就可以像普通列一样使用也可以在时间相关的操作中使用。
4.SQL 事件时间案例
来看看 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. 分配 watermarkDataStreamRow r env.addSource(new UserDefinedSource()).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractorRow(Time.minutes(0L)) {Overridepublic 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 中使用 f2String 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 SourceFunctionRow, ResultTypeQueryableRow {private volatile boolean isCancel;Overridepublic void run(SourceContextRow sourceContext) throws Exception {int i 0;while (!this.isCancel) {sourceContext.collect(Row.of(a i, b, System.currentTimeMillis()));Thread.sleep(10L);i;}}Overridepublic void cancel() {this.isCancel true;}Overridepublic TypeInformationRow getProducedType() {return new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),TypeInformation.of(Long.class));}}
}5.SQL 处理时间案例
来看看 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. 分配 watermarkDataStreamRow 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 中使用 f2String 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 SourceFunctionRow, ResultTypeQueryableRow {private volatile boolean isCancel;Overridepublic void run(SourceContextRow sourceContext) throws Exception {int i 0;while (!this.isCancel) {sourceContext.collect(Row.of(a i, b, System.currentTimeMillis()));Thread.sleep(10L);i;}}Overridepublic void cancel() {this.isCancel true;}Overridepublic TypeInformationRow getProducedType() {return new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),TypeInformation.of(Long.class));}}
}
