Apache Flink：如何使用 Table API 对每 n 行进行分组？

Question

最近我正在尝试使用 Apache Flink 进行快速批处理。 我有一个带有 column:value 和不相关索引列的表

基本上我想计算每 5 行值的平均值和范围。然后我将根据我刚刚计算的平均值计算平均值和标准差。所以我想最好的方法是使用Tumble窗口。

看起来像这样

DataSet<Tuple2<Double, Integer>> rawData = {get the source data};
Table table = tableEnvironment.fromDataSet(rawData);
Table groupedTable = table
            .window(Tumble.over("5.rows").on({what should I write?}).as("w")
            .groupBy("w")
            .select("f0.avg, f0.max-f0.min");

{The next step is to use groupedTable to calculate overall mean and stdDev} 

但我不知道在.on() 中写什么。我试过"proctime"，但它说没有这样的输入。我只希望它按从源读取的顺序分组。但它必须是时间属性，所以我不能使用"f2" - 索引列也是排序。

我是否必须添加时间戳才能执行此操作？批处理中是否有必要，它会减慢计算速度吗？解决这个问题的最佳方法是什么？

更新： 我尝试在表格 API 中使用滑动窗口，但它得到了异常。

// Calculate mean value in each group
    Table groupedTable = table
            .groupBy("f0")
            .select("f0.cast(LONG) as groupNum, f1.avg as avg")
            .orderBy("groupNum");

//Calculate moving range of group Mean using sliding window
    Table movingRangeTable = groupedTable
            .window(Slide.over("2.rows").every("1.rows").on("groupNum").as("w"))
            .groupBy("w")
            .select("groupNum.max as groupNumB, (avg.max - avg.min) as MR");

例外是：

线程“main”java.lang.UnsupportedOperationException 中的异常：当前不支持在事件时间计算滑动组窗口。

在 org.apache.flink.table.plan.nodes.dataset.DataSetWindowAggregate.createEventTimeSlidingWindowDataSet(DataSetWindowAggregate.scala:456)

在 org.apache.flink.table.plan.nodes.dataset.DataSetWindowAggregate.translateToPlan(DataSetWindowAggregate.scala:139)

...

这是否意味着 Table API 不支持滑动窗口？如果我没记错的话，DataSet API 中没有窗口函数。那么如何在批处理中计算移动范围呢？

Answer 1

window 子句用于定义基于窗口函数的分组，例如Tumble 或Session。除非您指定行的顺序，否则在 Table API（或 SQL）中没有很好地定义每 5 行分组。这是在Tumble 函数的on 子句中完成的。由于此功能源自流处理，on 子句需要时间戳属性。

您可以使用currentTimestamp() 函数获取当前时间的时间戳。但是，我应该指出，Flink 会对数据进行排序，因为它不知道函数的单调性。此外，所有这些都将具有 1 的并行度，因为没有允许分区的子句。

或者，您还可以实现一个用户定义的标量函数，将索引属性转换为时间戳（实际上是一个 Long 值）。但同样，Flink 会对数据进行完整的排序。