Flink: не найден подходящий метод для процесса

Question

Я пытаюсь применить функциональность для каждого окна в DataStream Флинка.Ниже приведен мой код

DataStream<Tuple2<String, String>> data = ...
DataStream<Tuple2<String, String>> freqCityChangeTransactions = data
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .process(new MyProcessWindowFunction());

Ниже приведена моя реализация MyProcessWindowFunction

public static class MyProcessWindowFunction
extends ProcessWindowFunction<Tuple2<String, String>, Tuple2<String, String>, String, TimeWindow> {

public void process(String key,
            Context context,
            Iterable<Tuple2<String, String>> input,
            Collector<Tuple2<String, String>> out) {
        // Do something ...
    }
}

Однако, когда я пытаюсь скомпилировать приведенный выше код через maven, я получаю следующую ошибку

[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.6.1:compile (default-compile) on project flink-examples: Compilation failure
[ERROR] /Users/furqan/Workspace/flink/src/main/java/com/baig/bank/Bank.java:[120,13] no suitable method found for process(com.baig.Bank.MyProcessWindowFunction)
[ERROR] method org.apache.flink.streaming.api.datastream.WindowedStream.<R>process(org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction<org.apache.flink.api.java.tuple.Tuple2<java.lang.String,java.lang.String>,R,org.apache.flink.api.java.tuple.Tuple,org.apache.flink.streaming.api.windowing.windows.TimeWindow>) is not applicable
[ERROR] (cannot infer type-variable(s) R
[ERROR] (argument mismatch; com.baig.Bank.MyProcessWindowFunction cannot be converted to org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction<org.apache.flink.api.java.tuple.Tuple2<java.lang.String,java.lang.String>,R,org.apache.flink.api.java.tuple.Tuple,org.apache.flink.streaming.api.windowing.windows.TimeWindow>))
[ERROR] method org.apache.flink.streaming.api.datastream.WindowedStream.<R>process(org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction<org.apache.flink.api.java.tuple.Tuple2<java.lang.String,java.lang.String>,R,org.apache.flink.api.java.tuple.Tuple,org.apache.flink.streaming.api.windowing.windows.TimeWindow>,org.apache.flink.api.common.typeinfo.TypeInformation<R>) is not applicable
[ERROR] (cannot infer type-variable(s) R
[ERROR] (actual and formal argument lists differ in length))

Есть идеи, что я тут не так делаю?К вашему сведению, я работаю с Apache Flink версии 1.5.1 и компилирую код Java с помощью maven3 на Mac.

Answer 1

Проблема в том, что существует несоответствие между KeySelector, используемым в keyBy, и типом ключа, указанным в ProcessWindowFunction.Вы указали ключ, используя индекс в Tuple2, и, как следствие, компилятор не может сделать вывод, что ключи будут Strings.В этой ситуации Flink передает ключ как Tuple.

Существует несколько способов исправить это.Если вы оставите keyBy как есть, вам нужно будет изменить ProcessWindowFunction, чтобы использовать Tuple в качестве типа ключа, и вам придется преобразовать ключ в String, если вы хотите его использовать.Что-то вроде ((Tuple1<String>)key).f0.Лучшим решением было бы использование более явного селектора ключей, такого как keyBy(t -> t.f0), так что ключи, как известно, являются строками во время компиляции.