Flink：找不到适合进程的方法

Question

我正在尝试在 Flink 的 DataStream 上应用每个 window 功能。以下是我的代码

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 版并在 Mac.

上使用 maven3 编译 Java 代码

Answer 1

问题是 keyBy 中使用的 KeySelector 与 ProcessWindowFunction 中指定的键类型不匹配。您已使用 Tuple2 中的索引指定键，因此，编译器无法推断键将是字符串。在这种情况下，Flink 将密钥作为元组传递。

有几种方法可以解决此问题。如果按原样保留 keyBy，则需要修改 ProcessWindowFunction 以使用 Tuple 作为键类型，如果要使用它，则必须将键转换为 String。类似于 ((Tuple1<String>)key).f0。更好的解决方案是使用更明确的键选择器，例如 keyBy(t -> t.f0)，以便在编译时知道键是字符串。

Flink：找不到适合进程的方法

Flink: no suitable method found for process

java

maven-3

apache-flink