您可以使用泛型类型实现Flink的AggregateFunction吗？

如何解决您可以使用泛型类型实现Flink的AggregateFunction吗？

我的目标是为Flink 1.10中的流处理模块提供一个接口。管道在其他运算符中包含一个AggregateFunction。所有运算符都具有通用类型，但是问题出在AggregateFunction内，该函数无法确定输出类型。

注意：实际的管道有一个slideEventEventWindow分配器和一个与AggregateFunction一起传递的WindowFunction，但是使用下面的代码可以更容易地再现该错误。

这是一个重现该错误的简单测试用例：

    @Test
    public void aggregateFunction_genericType() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Tuple2<String,Integer>> source = env.fromElements(Tuple2.of("0",1),Tuple2.of("0",2),3));

        ConfigAPI cfg = new ConfigAPI();

        source
                .keyBy(k -> k.f0)
                .countWindow(5,1)
                .aggregate(new GenericAggregateFunc<>(cfg))
                .print();


        env.execute();
    }

如您所见，Configuration类作为参数传递给Custom AggregationFunction。这就是用户要实现的。

    public static class ConfigAPI implements BaseConfigAPI<Tuple2<String,Integer>,Tuple2<String,Integer>> {
        @Override
        public Tuple2<String,Integer> createAcc() {
            return new Tuple2<>("0",0);
        }

        @Override
        public Tuple2<String,Integer> addAccumulators(Tuple2<String,Integer> in,Integer> acc) {
            acc.f1 += in.f1;
            return acc;
        }
    }

提供的接口是：

    public interface BaseConfigAPI<In,Acc> {
        Acc createAcc();
        Acc addAccumulators(In in,Acc acc);
        // other methods to override
    }

GenericAggregateFunction：

    public static class GenericAggregateFunc<In,Acc> implements AggregateFunction<In,Acc,Acc> {

        private BaseConfigAPI<In,Acc> cfg;
        GenericAggregateFunc(BaseConfigAPI<In,Acc> cfg) {
            this.cfg = cfg;
        }
        @Override
        public Acc createAccumulator() {
            return cfg.createAcc();
        }
        @Override
        public Acc add(In in,Acc acc) {
            return cfg.addAccumulators(in,acc);
        }
        @Override
        public Acc getResult(Acc acc) {
            return acc;
        }
        @Override
        public Acc merge(Acc acc,Acc acc1) {
            return null;
        }
    }

输出日志：

org.apache.flink.api.common.functions.InvalidTypesException: 
Type of TypeVariable 'Acc' in 'class misc.SlidingWindow$GenericAggregateFunc' could not be determined. This is most likely a type erasure problem. 
The type extraction currently supports types with generic variables only in cases where all variables in the return type can be deduced from the input type(s). 
Otherwise the type has to be specified explicitly using type information.

解决方案1（不起作用）：起初我以为这是“无法确定返回类型”的常见情况，所以我尝试添加

.returns(Types.TUPLE(Types.STRING,Types.INT))在.aggregate(...)之后，但没有成功。

解决方案2（有效）：我创建了一个具有通用类型的Wrapper类，名为Accumulator<Acc>，然后将其作为Type传递给 AggregateFunction<In,Accumulator<Acc>,Accumulator<Acc>>似乎可以正常工作。

这看起来不太优雅，并且与其余界面也不十分一致。还有其他解决方案吗？

编辑：感谢@deduper的时间和见解，我想我找到了解决方法。

解决方案3（有效）：我创建了一个新界面，该界面以以下方式扩展了BaseConfigAPI和AggregateFunction：

public interface MergedConfigAPI<In,Out> extends BaseConfigAPI,AggregateFunction<In,Out> {}

public interface BaseConfigAPI extends Serializable {
    //These will be implemented directly from AggregateFunction interface
    //Acc createAcc();
    //Acc addAccumulators(In in,Acc acc);
        
    //other methods to override
}

现在，用户只需实现MergedConfigAPI<In,Out>并将其作为参数传递给.aggregate(...)函数。

更新：我针对该框架测试了@deduper的第3个解决方案，该解决方案也不起作用。似乎由Acc而不是Out类型引发了异常。仔细研究.aggregate运算符的内部结构，我意识到有一个重载的aggregate方法需要另外2个参数。 TypeInformation<ACC> accumulatorType和TypeInformation<R> returnType。

这是最简单的解决方案在没有任何代码重构的情况下出现的方式。

解决方案4（有效）：

 @Test
 public void aggregateFunction_genericType() throws Exception {
                ...

                .aggregate(
                        new GenericAggregateFunc<>(cfg),Types.TUPLE(Types.STRING,Types.INT),Types.INT))
                ...
    }

注意：从Flink 1.10.1开始，aggregate方法使用@PublicEvolving进行注释。

解决方法

„ 您可以用泛型类型实现Flink的AggregateFunction吗？”

是的。您可以。由于您已经做好了自己。您的错误是由于您如何使用（如“ 使用站点通用名称”）而不是实现。

„ ...还有其他解决方案吗？... ”

我以简单…

的升序提出以下三种候选解决方案

...
source
       .keyBy(k -> k.f0)
       .countWindow(5,1)
       .aggregate(new GenericAggregateFunc< Tuple2<String,Integer>,Tuple2<String,Integer> >(cfg)) /* filling in the diamond will aid type inference */
       .print();
...

以上是最简单的，因为您不必重构原始的 GenericAgregateFunc ；只需使用要实例化泛型类的特定类型参数填充菱形。

还有另一种不太简单的解决方案……

public static class GenericAggregateFunc implements AggregateFunction<Tuple2<String,Integer>> {

    private BaseConfigAPI<Tuple2<String,Integer>> cfg;
    GenericAggregateFunc(BaseConfigAPI<Tuple2<String,Integer>> cfg) {
        this.cfg = cfg;
    }
    @Override
    public Tuple2<String,Integer> createAccumulator() {
        return cfg.createAcc();
    }
    @Override
    public Tuple2<String,Integer> add(Tuple2<String,Integer> in,Integer> acc) {
        return cfg.addAccumulators(in,acc);
    }
    @Override
    public Tuple2<String,Integer> getResult(Tuple2<String,Integer> acc) {
        return acc;
    }
    @Override
    public Tuple2<String,Integer> merge(Tuple2<String,Integer> acc,Integer> acc1) {
        return null;
    }
}

尽管此过程涉及较小的重构，但它会比第一个提出的解决方案（我认为）更简化整个应用程序。

Flink已经为您处理了“ 复杂”通用多态性。要插入Flink的所有您只需实例化其内置的通用 AggregateFunction<IN,ACC,OUT> ，并使用特定的输入要实例化的参数。在您的情况下，这些类型参数的类型为 Tuple2<String,Integer> 。

因此，第二种解决方案仍然是“ 使用泛型”，但是您使用的方式要简单得多。

另一个选项更接近您的原始实现，但有一些小的重构……

public static class GenericAggregateFunc<In,Acc,Out> implements AggregateFunction<In,Out> {
    
    ...
    @Override
    public Out getResult(Acc acc) {
        return ...;
    }
    ...
}

此外，要强制，前提是用户的配置必须实现与您的功能兼容的接口...

public interface BaseConfigAPI< In,Out >{ ... }

在my experiment中，我已经确认将 Out 类型参数也添加到 BaseConfigAPI 中，使其兼容。 / p>

我确实有一个更复杂的解决方案。但是由于越简单总是越好，所以我将把更复杂的解决方案留给别人提出。

您可以使用泛型类型实现Flink的AggregateFunction吗？

如何解决您可以使用泛型类型实现Flink的AggregateFunction吗？

解决方法

相关推荐