Flink SQL 已成为低代码流分析的标准，并设法统一批处理和流处理，同时保持 SQL 标准。此外，它还提供了一组丰富的实时数据分析高级功能。简而言之，Flink SQL 是两全其美的：它使您能够使用 SQL 处理流数据，但它还支持批处理。

Ververica Platform 使 Flink SQL 更易于跨团队访问和高效扩展。该平台附带了用于开发 SQL 脚本、管理用户定义函数 (UDF)、目录和连接器以及操作生成的长时间运行查询的附加工具。

我们已经看到了 Flink SQL 的许多用例，我们很高兴向您展示你可以用它构建什么。在本系列博文中，我们将探讨如何使用 Flink SQL 以多种方式处理数据。这篇文章将特别关注两个查询：Window Top-N 和 Continuous Top-N。

提示：访问我们的案例研究，探索其他人如何使用 Apache Flink。

什么是 Window Top-N 和 Continuous Top-N 查询？

Window Top-N 和 Continuous Top-N 是两种相似但略有不同的数据处理方式。在这两种情况下，我们都希望找到数据流中的前 N 项，但存在一些关键差异：

在 Window Top-N 中，我们在固定大小的窗口中处理数据。例如，我们可能希望每分钟找到前 10 个项目。

在 Continuous Top-N 中，我们连续处理数据。我们不使用窗口，而是在数据到达时对其进行处理。连续 Top-N 比窗口 Top-N 更难实现，但它有一些优点。例如，它可以更快地为我们提供结果，因为我们不必等待窗口关闭才能看到结果。

窗口 Top-N 和连续 Top-N 查询的常见用例

窗口 Top-N 和连续 Top-N 查询对于各种任务都很有用。例如，它们可以用于：

• 欺诈检测：在金融交易流中，我们可能希望找到每分钟按金额排名前 10 的交易。它可以帮助我们识别可疑活动。
• 用户交互流：在用户交互流中，我们可能希望找到正在查看或购买的前 10 件商品。它可以帮助我们向用户提出建议。
• 异常检测：在传感器读数流中，我们可能希望找到读数最高的前 10 个传感器。它可以帮助我们识别监控中出现故障的传感器。
• 日志消息流：在日志消息流中，我们可能希望按数量查找前 10 条日志消息。它可以帮助我们识别系统问题。

如何使用 Flink SQL 编写 Window Top-N 查询

首先我们来看看如何使用 Flink SQL 编写 Window Top-N 查询。我们将向您展示如何计算每个翻滚 5 分钟窗口中销售额最高的前 3 名供应商。

CREATE TABLE orders (
  bidtime TIMESTAMP(3),
  price DOUBLE,
  item STRING,
  supplier STRING,
  WATERMARK FOR bidtime AS bidtime - INTERVAL '5' SECONDS
) WITH (
  'connector' = 'faker',
  'fields.bidtime.expression' = '#{date.past ''30'',''SECONDS''}',
  'fields.price.expression' = '#{Number.randomDouble ''2'',''1'',''150''}',
  'fields.item.expression' = '#{Commerce.productName}',
  'fields.supplier.expression' = '#{regexify ''(Alice|Bob|Carol|Alex|Joe|James|Jane|Jack)''}',
  'rows-per-second' = '100'
);

SELECT *
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
  FROM (
    SELECT window_start, window_end, supplier, SUM(price) as price, COUNT(*) as cnt
    FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(bidtime), INTERVAL '5' MINUTE))
    GROUP BY window_start, window_end, supplier
  )
) WHERE rownum <= 3;

源表（orders）由 thefaker 连接器支持，该连接器根据 Java Faker 表达式在内存中不断生成行。注意：此示例利用 Window Top-N 功能来显示排名前 3 的供应商每 5 分钟最高销售额。

如何使用 Flink SQL 编写连续 Top-N 查询

编写连续 Top-N 查询比编写 Window Top-N 查询更困难。这样做的原因是，在 Continuous Top-N 中，我们在数据到达时对其进行处理，而不是使用窗口。这个示例将带我们进入神奇的领域，因为流处理通常被外行认为是这样。然而，它实际上只是在数据流上执行的一组指令。

我们将展示如何使用 OVER window 和 ROW_NUMBER() 函数根据给定属性连续计算“Top-N”行。源表 (spells_cast) 由 thefaker 连接器支持，该连接器基于 Java Faker 在内存中连续生成行。

CREATE TABLE spells_cast (
  wizard STRING,
  spell STRING
) WITH (
  'connector' = 'faker',
  'fields.wizard.expression' = '#{harry_potter.characters}',
  'fields.spell.expression' = '#{harry_potter.spells}'
);

SELECT wizard, spell, COUNT(*) AS times_cast
FROM spells_cast
GROUP BY wizard, spell;

此结果可以在 OVER 窗口中用于计算 Top-N。使用 wizard 列对行进行分区，然后根据施法次数（times_cast DESC）进行排序。内置函数 ROW_NUMBER() 根据排序顺序为每个分区的行分配序号。通过筛选序号小于等于 N 的行，我们可以获得每个巫师施法次数前 N 的法术。

以下是一个示例查询：

SELECT wizard, spell, times_cast
FROM (
  SELECT wizard, spell, times_cast,
         ROW_NUMBER() OVER (PARTITION BY wizard ORDER BY times_cast DESC) AS row_num
  FROM (
    SELECT wizard, spell, COUNT(*) AS times_cast
    FROM spells_cast
    GROUP BY wizard, spell
  )
)
WHERE row_num <= 3;

在此查询中，我们首先计算每个巫师施法次数的统计信息。然后，我们在内部查询中使用 ROW_NUMBER() 函数对每个巫师的法术按照施法次数进行降序排列，为每个法术分配行号。最后，在外部查询中，我们筛选出行号小于等于 3 的记录，以获取每个巫师施法次数前 3 的法术。

这就是如何使用 Flink SQL 编写连续 Top-N 查询的方式。通过以上方法，您可以处理实时数据流并获取持续更新的 Top-N 数据。

请注意，上述示例中的 SQL 查询是根据您提供的上下文进行的翻译和还原，可能会因为特定上下文的变化而略有不同。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典 ，回复“大数据”获取多本大数据电子书