Apache Flink 两种类型的窗口，时间和“计数”窗口

Posted 2023-04-18

【中文标题】Apache Flink 两种类型的窗口，时间和“计数”窗口

【英文标题】：Apache Flink two types of window, time and "count" windows

【发布时间】：2018-06-23 15:27:49

【问题描述】：

我正在尝试将文件作为带有窗口的流来处理。

这里是代码

object Prog 

  def main(args: Array[String]) : Unit = 
    org.apache.log4j.BasicConfigurator.configure()

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = TableEnvironment.getTableEnvironment(env)

    val csvTableSource = CsvTableSource
      .builder
      .path("src/main/resources/data.stream")
      .field("numPers", Types.INT)
      .field("TIMESTAMP", Types.STRING)
      .fieldDelimiter(",")
      .ignoreFirstLine
      .ignoreParseErrors
      .commentPrefix("%")
      .build()

    tableEnv.registerTableSource("Data", csvTableSource)

    val table = tableEnv.scan("Data")
      .filter("numPers > 10")
      .select("*")

    val ds = tableEnv.toAppendStream(table, classOf[Row])

    ds.print()
    env.execute()
  

问题是如何在这里实现窗口，例如，只显示不超过一小时的值。 或者第二种窗口类型，当我读到最后 50 个条目时。

【参考方案1】：

在流处理中，窗口是计算聚合的组。

您的用例似乎有所不同。如果您想保留最后的 x 分钟或 y 最后的记录，则需要在 SQL 中以不同的方式表示。

保留最后的5 分钟类似于

SELECT * FROM Data d WHERE d.tstamp >  (now() - INTERVAL '5' MINUTE)

所以，这将是某种时间戳属性的过滤器。

保留最后 10 行将是

SELECT * FROM Data d ORDER BY d.tstamp DESC LIMIT 10

但是，Flink（1.5 版）SQL 或 Table API 尚不支持这些操作。