Не удается разрешить метод уменьшения при использовании Apache Flink

Question

Я следую документам из Apache Флинк для сбора данных, это мой код:

     consumeRecord.keyBy("gameType")
                    .timeWindow(Time.seconds(5))
                    .reduce(new MyReduceFunction(),new MyProcessWindowFunction());

 private static class MyReduceFunction implements ReduceFunction<ReportPump> {
        public ReportPump reduce(ReportPump d1, ReportPump d2) {
            d1.setRealPumpAmount(d1.getRealPumpAmount() + d2.getRealPumpAmount());
            d1.setPumpAmount(d1.getPumpAmount() + d2.getPumpAmount());
            return d1;
        }
    }

    private static class MyProcessWindowFunction extends ProcessWindowFunction<ReportPump, Tuple2<Long, ReportPump>, String, TimeWindow> {
        public void process(String key,
                            Context context,
                            Iterable<ReportPump> minReadings,
                            Collector<Tuple2<Long, ReportPump>> out) {
            ReportPump min = minReadings.iterator().next();
            out.collect(new Tuple2<Long, ReportPump>(context.window().getStart(), min));
        }
    }

, но код говорит мне:

Cannot resolve method 'reduce(com.sportswin.rdw.PumpRealtimeHandler.MyReduceFunction, com.sportswin.rdw.PumpRealtimeHandler.MyProcessWindowFunction)'

что я должен сделать, чтобы решить эту проблему? ReportPump - это объект, который я определил.

Answer 1

Если вы .keyBy("fieldName") Flink не может определить тип ключа, который вы запрашиваете. Вот почему предполагается, что в качестве типа ключа используется generic c type Tuple, и ваша функция обработки должна расширяться ProcessWindowFunction<ReportPump, Tuple2<Long, ReportPump>, Tuple, TimeWindow>.

Если вы вместо этого хотите использовать поле keyBy gameType в ваших данных с String введите, что вы можете позвонить что-то вроде:

.keyBy(pump -> pump.getGameType() // I assume this has a String type

и тогда ваш код скомпилируется.