Flink：找不到合适的工艺方法

Question

我试图在Flink的数据流上应用每窗口功能。以下是我的代码

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

下面是我的实现MyProcessWindow函数

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))

知道我做错了什么吗？仅供参考，我正在使用ApacheFlink 1.5.1版，并在Mac上使用maven3编译Java代码。

Answer 1

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

有几种方法可以解决这个问题。如果您保持keyBy不变，那么您需要修改ProcessWindow函数以使用元组作为键类型，如果您想使用它，您必须将键转换为字符串。类似于（（Tuple1