文章目录

• Flink 系列文章
• 一、TOP-N

• 1、No Ranking Output Optimization（无排名输出优化）

• 二、Window Top-N

• 1、Window Top-N 语法
• 2、Window Top-N follows after Window Aggregation
• 3、Window Top-N follows after Windowing TVF
• 3、Limitation

• 三、Window Deduplication

• 1、语法
• 2、示例
• 3、限制

• 1）、Limitation on Window Deduplication which follows after Windowing TVFs directly
• 2）、Limitation on time attribute of order key

本文介绍了Flink 的Top-N 、window Top-N以及窗口去重 及具体的运行示例。
本文依赖flink和kafka集群能正常使用。
本文分为3个部分，即top-n、window top-n和window去重，并且每个内容均以验证通过示例进行说明。
本文的示例是在Flink 1.17版本上验证通过的。

前 N 个查询要求按列排序的 N 个最小值或最大值。最小值和最大值集都被视为前 N 个查询。在需要仅显示条件上批处理/流式处理表中最底部的 N 条记录或 N 条最前的记录的情况下，前 N 个查询非常有用。此结果集可用于进一步分析。

Flink 使用 OVER 窗口子句和过滤条件的组合来表达 Top-N 查询。借助 OVER window PARTITION BY 子句的强大功能，Flink 还支持每组 Top-N。例如，每个类别实时销售额最高的前五种产品。批处理表和流式处理表上的 SQL 支持前 N 个查询。

下面显示了 Top-N 语句的语法：

参数说明：

• ROW_NUMBER（）：根据分区中行的顺序，为每一行分配一个唯一的序列号，从 1 开始。目前（截至版本Flink 1.17），我们仅支持ROW_NUMBER作为窗口功能。将来，我们将支持 RANK（） 和 DENSE_RANK（）。
• PARTITION BY col1[， col2…]：指定分区列。每个分区都有一个前 N 个结果。
• ORDER BY col1 [asc|desc][， col2 [asc|desc]…]：指定排序列。不同列的排序方向可能不同。
• WHERE rownum <= N：Flink 需要 rownum <= N 才能识别此查询是 Top-N 查询。N 表示将保留的 N 条最小或最大记录。
• [AND 条件]：可以在 where 子句中添加其他条件，但其他条件只能使用 AND 连词与 rownum <= N 组合。

必须严格遵循上述模式，否则优化程序将无法转换查询。

TopN 查询是“结果更新”。Flink SQL 会根据顺序键对输入数据流进行排序，所以如果前 N 条记录被更改，更改后的记录将作为撤回/更新记录发送到下游。建议使用支持更新的存储作为 Top-N 查询的接收器。此外，如果需要将前 N 条记录存储在外部存储中，则结果表应与 Top-N 查询具有相同的唯一键。

Top-N 查询的唯一键是分区列和 rownum 列的组合。Top-N 查询还可以导出上游的唯一键。以下面的作业为例，假设product_id是 ShopSales 的唯一键，那么 Top-N 查询的唯一键是 [category， rownum] 和 [product_id]。

以下示例演示如何在流式处理表上使用 Top-N 指定 SQL 查询。这是一个示例，用于获取我们上面提到的“每个类别实时销售额最高的前五种产品”。

如上所述，rownum 字段将作为唯一键的一个字段写入结果表中，这可能会导致大量记录被写入结果表。例如，当排名 9 的记录（例如 product-1001）更新且其排名升级到 1 时，排名 1 ~ 9 的所有记录将作为更新消息输出到结果表中。如果结果表接收的数据过多，就会成为SQL作业的瓶颈。

优化方法是省略 Top-N 查询的外部 SELECT 子句中的 rownum 字段。这是合理的，因为前 N 条记录的数量通常不大，因此消费者可以快速对记录进行排序。如果没有 rownum 字段，在上面的例子中，只需要将更改的记录 （product-1001） 发送到下游，这可以减少结果表的大量 IO。

以下示例演示如何以这种方式优化上述 Top-N 示例：
本示例是基于上述示例中的表结构、数据，比较二者的查询结果。

为了将上述查询输出到外部存储并获得正确的结果，外部存储必须与 Top-N 查询具有相同的唯一键。在上面的示例查询中，如果product_id是查询的唯一键，则外部表也应具有product_id作为唯一键。

Window Top-N 是一个特殊的 Top-N，它返回每个窗口和其他分区键的 N 个最小值或最大值。

对于流式处理查询，与连续表上的常规 Top-N 不同，窗口 Top-N 不会发出中间结果，而只会发出最终结果，即窗口末尾的前 N 条记录总数。此外，窗口 Top-N 在不再需要时会清除所有中间状态。因此，如果用户不需要按记录更新结果，则窗口 Top-N 查询具有更好的性能。通常，Window Top-N直接与Windowing TVF一起使用。此外，Window Top-N可以与基于窗口TVF的其他操作一起使用，例如窗口聚合，窗口TopN和窗口联接。

Window Top-N can be defined in the same syntax as regular Top-N, see Top-N documentation for more information. Besides that, Window Top-N requires the PARTITION BY clause contains window_start and window_end columns of the relation applied Windowing TVF or Window Aggregation. Otherwise, the optimizer won’t be able to translate the query.
可以使用与常规 Top-N 相同的语法定义窗口 Top-N。除此之外，Window Top-N 要求 PARTITION BY 子句包含应用窗口 TVF 或窗口聚合的关系的window_start和window_end列。否则，优化程序将无法转换查询。

The following shows the syntax of the Window Top-N statement:

以下示例演示如何计算每个滚动 5 分钟窗口内销售额最高的前 3 个订单。

以下示例显示如何计算每个滚动 10 分钟窗口价格最高的前 3 个订单。

截至版本Flink 1.17，Flink 仅支持具有 Tumble Windows、Hop Windows 和 Cumulate Windows 的 Windowing TVF 的 Window Top-N 。Window Top-N 的 Windowing TVF 与会话窗口将在不久的将来得到支持。

窗口重复数据删除是一种特殊的重复数据删除，可删除在一组列上重复的行，为每个窗口和分区键保留第一个或最后一个行。

对于流式处理查询，与连续表上的常规重复数据删除不同，窗口重复数据删除不会发出中间结果，而只会在窗口结束时发出最终结果。此外，窗口重复数据删除会在不再需要时清除所有中间状态。因此，如果用户不需要更新每条记录的结果，则窗口重复数据删除查询具有更好的性能。通常，窗口重复数据删除直接与窗口化 TVF 一起使用。此外，窗口重复数据删除还可以与基于窗口化TVF的其他操作一起使用，例如窗口聚合，窗口TopN和窗口联接。

窗口重复数据删除可以使用与常规重复数据删除相同的语法进行定义。除此之外，窗口重复数据删除要求 PARTITION BY 子句包含关系的window_start和window_end列。否则，优化程序将无法转换查询。

Flink 使用 ROW_NUMBER（） 来删除重复项，就像 Window Top-N 查询的方式一样。理论上，窗口重复数据删除是窗口顶部 N 的一个特例，其中 N 是 1，并按处理时间或事件时间排序。

下面显示了窗口重复数据删除语句的语法：

必须完全遵循上述模式，否则优化程序不会将查询转换为窗口重复数据删除。

下面的示例演示如何保留每 10 分钟滚动窗口的最后一条记录。

1）、Limitation on Window Deduplication which follows after Windowing TVFs directly

截至 Flink 1.17版本，如果窗口重复数据删除在窗口化 TVF 之后进行，则窗口化 TVF 必须与 Tumble Windows、Hop Windows 或Cumulate Windows一起使用，而不是会话窗口。在不久的将来将支持会话窗口。

2）、Limitation on time attribute of order key

截至 Flink 1.17版本，窗口重复数据删除要求订单键必须是事件时间属性，而不是处理时间属性。在不久的将来将支持按处理时间排序。

以上，介绍了Flink 的Top-N 、window Top-N以及窗口去重 及具体的运行示例。