Flink DataSet API: Is GroupBy 无法正常工作

Question

在我的Flink Java程序中，我使用GroupBy-Operator，如下所示：

dataSet.groupBy(new KeySelector<myObject, Tuple2<Tuple2<Integer, Integer>, Integer>>() {
    private static final long serialVersionUID = 5L;
    Tuple2<Tuple2<Integer, Integer>, Integer> groupingKey = new Tuple2<Tuple2<Integer, Integer>, Integer>();
        public Tuple2<Tuple2<Integer, Integer>, Integer> getKey(myObject s) {
            groupingKey.setField(s.getPosition(), 0);
            groupingKey.setField(s.getBand(), 1);
            return groupingKey;
        }
    })
    .reduceGroup(reduceFunction);

getPosition()返回一个Tuple2<Integer, Integer>，getBand()返回一个int。

我想根据这两个值对数据集进行分组。如果我有 6 个位置和 4 个波段，我想得到 24 个不同的组，并为每个组独立使用 groupReduce -函数。但目前我生成的组似乎包含波段和位置的各种值。我在groupReduce函数中像这样检查了这一点：

if (this.band == null) {
    this.band = myObject.getBand();
}
if (this.band != myObject.getBand()) {
    System.out.println("The band should be " + this.band + " but is: " + myObject.getBand());

此外，我生成的文件中还有一些值表明分组有问题。在我的情况下，分组是否可能不起作用？或者这可能只是我的代码中另一个潜在错误的结果？

Answer 1

我认为您在GroupReduceFunction中的检查无法正常工作。对于不同的组，可以多次调用GroupReduceFunction.reduce()。 this.band是您GroupReduceFunction的成员变量，我假设您不会在reduce()方法结束时重置this.band。

因此，this.band只会在第一次调用reduce()时null。在第二次呼叫开始时，this.band将被初始化，并且不会设置为当前组的频段。因此，以下检查将失败。