Apache FlinkでAllWindowedStreamを使った処理を実装しようとしているのですが、下記のエラーの対処がわかりません。
[error] no suitable method found for process(TestProcessFunction)
[error] method org.apache.flink.streaming.api.datastream.AllWindowedStream.<R>process(org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction<org.apache.flink.api.java.tuple.Tuple4<java.lang.String,java.lang.String,java.lang.Integer,java.util.List<java.util.List<java.lang.Double>>>,R,org.apache.flink.streaming.api.windowing.windows.TimeWindow>) is not applicable
[error] (cannot infer type-variable(s) R
[error] (argument mismatch; HumanDetector.HumanDetector.TestProcessFunction cannot be converted to org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction<org.apache.flink.api.java.tuple.Tuple4<java.lang.String,java.lang.String,java.lang.Integer,java.util.List<java.util.List<java.lang.Double>>>,R,org.apache.flink.streaming.api.windowing.windows.TimeWindow>))
[error] method org.apache.flink.streaming.api.datastream.AllWindowedStream.<R>process(org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction<org.apache.flink.api.java.tuple.Tuple4<java.lang.String,java.lang.String,java.lang.Integer,java.util.List<java.util.List<java.lang.Double>>>,R,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))
[error] tuple_data
[error] .windowAll(SlidingProcessingTimeWindows.of(Time.milliseconds(100), Time.milliseconds(50)))
[error] .process
実装したコードを以下に示します。
DataStream:
DataStream<Tuple4<String, String, Integer, List<List<Double>>>> tuple_data = ...
DataStream<String> window_test = tuple_data
.windowAll(SlidingProcessingTimeWindows.of(Time.milliseconds(100), Time.milliseconds(50)))
.process(new TestProcessFunction());
Process Function:
public static class TestProcessFunction
extends ProcessAllWindowFunction<Tuple4<String, String, Integer, List<List<Double>>>, String, Window>{
@Override
public void process(Context context,
Iterable<Tuple4<String, String, Integer, List<List<Double>>>> input,
Collector<String> out) throws Exception {
long count = 0;
for (Tuple4<String, String, Integer, List<List<Double>>> in: input) {
count++;
}
out.collect("Window: " + context.window() + "count: " + count);
}
}
Apache Flinkのバージョンは1.14です。
よろしくお願いします。