ApacheFlink：如何使用表API对每n行进行分组？

Question

最近我尝试使用ApacheFlink进行快速批处理。我有一个表，它有一个列：value和一个不相关的索引列

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

看起来是这样的

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”，但它说没有这样的输入。我只希望它在从源代码读取时按顺序分组。但是它必须是一个时间属性，所以我不能使用索引列作为排序。

我是否必须添加时间戳才能执行此操作？在批处理过程中是否有必要这样做，是否会降低计算速度？解决这个问题的最好办法是什么？

更新：我试图在表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。阿帕奇。Flink。桌子计划节点。数据集。数据集设置为Waggreegate。createEventTimeSlidingWindowDataSet（DataSetWebindowaggregate.scala:456）

org.apache.flink.table.plan.nodes.dataset.DataSetWindowAggregate.translateToplan（DataSetWindowAggregate.scala:139）

...

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

Answer 1

window子句用于定义基于窗口函数的分组，例如翻滚或会话。表API（或SQL）中没有很好地定义每5行分组一次，除非指定行的顺序。这是在翻滚函数的on子句中完成的。由于此功能源自流处理，因此on子句需要一个timestamp属性。

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

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