- 921. Minimum Add to Make Parentheses Valid 使括号有效的最少添加
- 915. Partition Array into Disjoint Intervals 分割数组
- 932. Beautiful Array 漂亮数组
- 940. Distinct Subsequences II 不同的子序列 II
时间语义,要配合窗口操作才能发挥作用。最主要的用途,当然就是开窗口、根据时间 段做计算了。下面我们就来看看 Table API 和 SQL 中,怎么利用时间字段做窗口操作。
在Table API 和 SQL 中,主要有两种窗口:Group Windows 和 Over Windows
分组窗口(Group Windows)会根据时间或行计数间隔,将行聚合到有限的组(Group) 中,并对每个组的数据执行一次聚合函数。
Table API 中的 Group Windows 都是使用.window(w:GroupWindow)子句定义的,并且 必须由 as 子句指定一个别名。为了按窗口对表进行分组,窗口的别名必须在 group by 子句 中,像常规的分组字段一样引用。
Table table = input
.window([w: GroupWindow] as "w") // 定义窗口,别名 w
.groupBy("w, a") // 以属性 a 和窗口 w 作为分组的 key
.select("a, b.sum") // 聚合字段 b 的值,求和
或者,还可以把窗口的相关信息,作为字段添加到结果表中:
Table table = input
.window([w: GroupWindow] as "w")
.groupBy("w, a")
.select("a, w.start, w.end, w.rowtime, b.count")
Table API 提供了一组具有特定语义的预定义 Window 类,这些类会被转换为底层
DataStream 或 DataSet 的窗口操作。
Table API 支持的窗口定义,和我们熟悉的一样,主要也是三种:滚动(Tumbling)、滑 动(Sliding)和会话(Session)。
滚动窗口(Tumbling windows)要用 Tumble 类来定义,另外还有三个方法:
⚫over:定义窗口长度
⚫on:用来分组(按时间间隔)或者排序(按行数)的时间字段
⚫as:别名,必须出现在后面的 groupBy 中
代码如下:
// Tumbling Event-time Window
.window(Tumble.over("10.minutes").on("rowtime").as("w"))
// Tumbling Processing-time Window
.window(Tumble.over("10.minutes").on("proctime").as("w"))
// Tumbling Row-count Window
.window(Tumble.over("10.rows").on("proctime").as("w"))
滑动窗口(Sliding windows)要用 Slide 类来定义,另外还有四个方法:
⚫over:定义窗口长度
⚫every:定义滑动步长
⚫on:用来分组(按时间间隔)或者排序(按行数)的时间字段
⚫as:别名,必须出现在后面的 groupBy 中
代码如下:
// Sliding Event-time Window
.window(Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w"))
// Sliding Processing-time window
.window(Slide.over("10.minutes").every("5.minutes").on("proctime").as("w"))
// Sliding Row-count window
.window(Slide.over("10.rows").every("5.rows").on("proctime").as("w"))
会话窗口(Session windows)要用 Session 类来定义,另外还有三个方法:
⚫withGap:会话时间间隔
⚫on:用来分组(按时间间隔)或者排序(按行数)的时间字段
⚫as:别名,必须出现在后面的 groupBy 中
代码如下:
// Session Event-time Window
.window(Session.withGap.("10.minutes").on("rowtime").as("w"))
// Session Processing-time Window
.window(Session.withGap.("10.minutes").on(“proctime").as("w"))
Over window 聚合是标准 SQL 中已有的(Over 子句),可以在查询的 SELECT 子句中定义。Over window 聚合,会针对每个输入行,计算相邻行范围内的聚合。Over windows
使用.window(w:overwindows*)子句定义,并在 select()方法中通过别名来引用。
比如这样:
Table table = input
.window([w: OverWindow] as "w")
.select("a, b.sum over w, c.min over w")
Table API 提供了 Over 类,来配置 Over 窗口的属性。可以在事件时间或处理时间,以及 指定为时间间隔、或行计数的范围内,定义 Over windows。
无界的over window 是使用常量指定的。也就是说,时间间隔要指定 UNBOUNDED_RANGE, 或者行计数间隔要指定 UNBOUNDED_ROW。而有界的 over window 是用间隔的大小指定的。
实际代码应用如下:
1、 无界的overwindow;
// 无界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding.(UNBOUNDED_RANGE).as("w"))
// 无界的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding.(UNBOUNDED_RANGE).as("w"))
// 无界的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding.(UNBOUNDED_ROW). as("w"))
//无界的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("proctime").preceding.(UNBOUNDED_ROW).as("w"))
1、 有界的overwindow;
// 有界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("1.minutes").as("w"))
// 有界的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding("1.minutes").as ("w"))
// 有界的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("10.rows").as("w "))
// 有界的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("procime").preceding("10.rows").as("w "))
我们已经了解了在 Table API 里 window 的调用方式,同样,我们也可以在 SQL 中直接加入窗口的定义和使用。
Group Windows 在 SQL 查询的 Group BY 子句中定义。与使用常规 GROUP BY 子句的查询 一样,使用 GROUP BY 子句的查询会计算每个组的单个结果行。
SQL支持以下 Group 窗口函数:
•TUMBLE(time_attr, interval)
定义一个滚动窗口,第一个参数是时间字段,第二个参数是窗口长度。
•HOP(time_attr, interval, interval)
定义一个滑动窗口,第一个参数是时间字段,第二个参数是窗口滑动步长,第三个是窗 口长度。
•SESSION(time_attr, interval)
定义一个会话窗口,第一个参数是时间字段,第二个参数是窗口间隔(Gap)。
另外还有一些辅助函数,可以用来选择 Group Window 的开始和结束时间戳,以及时间 属性。
这里只写 TUMBLE_,滑动和会话窗口是类似的(HOP_,SESSION_*)。
•TUMBLE_START(time_attr, interval)
•TUMBLE_END(time_attr, interval)
•TUMBLE_ROWTIME(time_attr, interval)
•TUMBLE_PROCTIME(time_attr, interval)
由于 Over 本来就是 SQL 内置支持的语法,所以这在 SQL 中属于基本的聚合操作。所有 聚合必须在同一窗口上定义,也就是说,必须是相同的分区、排序和范围。目前仅支持在当 前行范围之前的窗口(无边界和有边界)。
注意,ORDER BY 必须在单一的时间属性上指定。
代码如下:
SELECT COUNT(amount) OVER ( PARTITION BY user
ORDER BY proctime
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
FROM Orders
// 也可以做多个聚合
SELECT COUNT(amount) OVER w, SUM(amount) OVER w FROM Orders
WINDOW w AS ( PARTITION BY user
ORDER BY proctime
代码:
package org.flink.tableapi;
import org.flink.beans.SensorReading;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
public class TableTest5_TimeAndWindow {
public static void main(String[] args) throws Exception {
// 1. 创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 2. 读入文件数据,得到DataStream
DataStream<String> inputStream = env.readTextFile("C:\\Users\\Administrator\\IdeaProjects\\FlinkStudy\\src\\main\\resources\\sensor.txt");
// 3. 转换成POJO
DataStream<SensorReading> dataStream = inputStream.map(line -> {
String[] fields = line.split(",");
return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
})
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
@Override
public long extractTimestamp(SensorReading element) {
return element.getTimestamp() * 1000L;
}
});
// 4. 将流转换成表,定义时间特性
// Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");
Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, rt.rowtime");
tableEnv.registerTable("sensor", dataTable);
// 5. 窗口操作
// 5.1 Group Window
// table API
Table resultTable = dataTable.window(Tumble.over("10.seconds").on("rt").as("tw"))
.groupBy("id, tw")
.select("id, id.count, temp.avg, tw.end");
// SQL
Table resultSqlTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp, tumble_end(rt, interval '10' second) " +
"from sensor group by id, tumble(rt, interval '10' second)");
// 5.2 Over Window
// table API
Table overResult = dataTable.window(Over.partitionBy("id").orderBy("rt").preceding("2.rows").as("ow"))
.select("id, rt, id.count over ow, temp.avg over ow");
// SQL
Table overSqlResult = tableEnv.sqlQuery("select id, rt, count(id) over ow, avg(temp) over ow " +
" from sensor " +
" window ow as (partition by id order by rt rows between 2 preceding and current row)");
// dataTable.printSchema();
// tableEnv.toAppendStream(resultTable, Row.class).print("result");
// tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");
tableEnv.toAppendStream(overResult, Row.class).print("result");
tableEnv.toRetractStream(overSqlResult, Row.class).print("sql");
env.execute();
}
}
测试记录:
我已经设置了 Azure API 管理服务,并在自定义域上配置了它。在 Azure 门户中 API 管理服务的配置部分下,我设置了以下内容: 因为这是一个客户端系统,我必须屏蔽细节,但以下是基础知识:
我是一名习惯 React Native 的新程序员。我最近开始学习 Fetch API 及其工作原理。我的问题是,我找不到人们使用 API key 在他们的获取语句中访问信息的示例(我很难清楚地表达有
这里有很多关于 API 是什么的东西,但是我找不到我需要的关于插件 API 和类库 API 之间的区别。反正我不明白。 在 Documenting APIs 一书中,我读到:插件 API 和类库 AP
关闭。这个问题不满足Stack Overflow guidelines .它目前不接受答案。 想改善这个问题吗?更新问题,使其成为 on-topic对于堆栈溢出。 7年前关闭。 Improve thi
我正在尝试找出设计以下场景的最佳方法。 假设我已经有了一个 REST API 实现,它将从不同的供应商那里获取书籍并将它们返回给我自己的客户端。 每个供应商都提供单独的 API 来向其消费者提供图书。
请有人向我解释如何使用 api key 以及它有什么用处。 我对此进行了很多搜索,但得到了不同且相互矛盾的答案。有人说 API key 是保密的,它从不作为通信的一部分发送,而其他人则将它发送给客户端
关闭。这个问题是opinion-based .它目前不接受答案。 想改进这个问题?更新问题,以便 editing this post 可以用事实和引用来回答它. 4年前关闭。 Improve this
谁能告诉我为什么 WSo2 API 管理器不进行身份验证?我已经设置了两个 WSo2 API Manager 1.8.0 实例并创建了一个 api。它作为原型(prototype) api 工作正常。
我在学习 DSL 的过程中遇到了 Fluent API。 我在流利的 API 上搜索了很多……我可以得出的基本结论是,流利的 API 使用方法链来使代码流利。 但我无法理解——在面向对象的语言中,我们
基本上,我感兴趣的是在多个区域设置 WSO2 API 管理器;例如亚洲、美国和欧洲。一些 API 将部署在每个区域的数据中心内,而其他 API 将仅部署在特定区域内。 理想情况下,我想要的是一个单一的
我正在构建自己的 API,供以下用户使用: 1) 安卓应用 2) 桌面应用 我的网址之一是:http://api.chatapp.info/order_api/files/getbeers.php我的
我需要向所有用户显示我的站点的分析,但使用 OAuth 它显示为登录用户配置的站点的分析。如何使用嵌入 API 实现仪表板但仅显示我的网站分析? 我能想到的最好的可能性是使用 API key 而不是客
我正在研究大公司如何管理其公共(public) API。我想到的是拥有成熟 API 的公司,例如 Google、Facebook、Twitter 和 Amazon。 这些公司向公众公开了许多不同的 A
在定义客户可访问的 API 时,以下是首选的行业惯例: a) 定义一组显式 API 方法,每个方法都有非常狭窄和特定的目的,例如: SetUserName SetUserAge Se
这在本地 deserver 和部署时都会发生。我成功地能够通过留言簿教程使用 API 资源管理器,但现在我已经创建了自己的项目并尝试访问我编写的第一个 API,它从未出现过。搜索栏旁边的黄色“正在加载
我正在尝试使用 http://ip-api.com/ api通过我的ip地址获取经度和纬度。当我访问 http://ip-api.com/json从我的浏览器或使用 curl,它以 json 格式返回
这里的典型示例是 Twitter 的 API。我从概念上理解 REST API 的工作原理,本质上它只是针对您的特定请求向他们的服务器查询,然后您会在其中收到响应(JSON、XML 等),很棒。 但是
我能想到的最好的标题,但要澄清的是,情况是这样的: 我正在开发一种类似短 url 的服务,该服务允许用户使用他们的 Twitter 帐户“登录”并发布内容。现在这项服务可以包含在 Tweetdeck
我正在设计用于管理评论和讨论线程的 API 方案。我想有一个点 /discussions/:discussionId 当您GET 时,它会返回一组评论和一些元数据。评论也许可以单独访问 /discus
关闭。这个问题需要更多focused .它目前不接受答案。 想改进这个问题吗? 更新问题,使其只关注一个问题 editing this post . 关闭去年。 Improve this quest
我是一名优秀的程序员,十分优秀!