Ошибки функции процесса Apache Flink ⇐ JAVA

[Гость]

I try to test simple Process Function of Apache Flink with java api.

IDE : Visual Studio code 1.87.1

Flink : 1.18.1

== CountWithTimestamp.java

public class CountWithTimestamp { public String key; public long count; public long lastModified; } == CountWithTimeoutFunction.java

import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.util.Collector; public class CountWithTimeoutFunction extends KeyedProcessFunction{ private ValueState state; @Override public void open(Configuration parameters) throws Exception { // TODO Auto-generated method stub state = this.getRuntimeContext().getState(new ValueStateDescriptor("My_State", CountWithTimestamp.class)); } @Override public void processElement(Tuple2 value, Context ctx, Collector out) throws Exception { // TODO Auto-generated method stub CountWithTimestamp current = state.value(); if (current == null) { current = new CountWithTimestamp(); current.key = value.f0; } current.count++; current.lastModified = ctx.timestamp(); state.update(current); ctx.timerService().registerEventTimeTimer(current.lastModified + 60000); } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception { // TODO Auto-generated method stub CountWithTimestamp result = state.value(); if (timestamp == result.lastModified + 60000) { out.collect(new Tuple2(result.key, result.count)); } } } == JavaFlinkTest.java

import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.typeinfo.TypeHint; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class JavaFlinkTest { public static void main( String[] args ) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream ds = env.fromElements( "To be, or not to be, that is the question:", "Whether 'tis nobler in the mind to suffer", "The slings and arrows of outrageous fortune", "Or to take arms against a sea of troubles," ).flatMap(new Splitter()) .keyBy(value -> value.f0) .process(new CountWithTimeoutFunction(), TypeInformation.of(new TypeHint(){})); --> This line throws Errors ds.print(); env.execute("Window WordCount"); env.close(); } public static class Splitter implements FlatMapFunction { @Override public void flatMap(String sentence, Collector out) throws Exception { for (String word: sentence.split(" ")) { out.collect(new Tuple2(word, 1L)); } } } } But the errors occur in KeyedStream.process method. And the error messages are,

The method process(ProcessFunction, TypeInformation) in the type KeyedStream is not applicable for the arguments (CountWithTimeoutFunction, TypeInformation)Java(67108979) Go to Super Implementation SingleOutputStreamOperator org.apache.flink.streaming.api.datastream.KeyedStream.process(ProcessFunction processFunction, TypeInformation outputType) Deprecated. Use KeyedStream.process(KeyedProcessFunction, TypeInformation) Applies the given ProcessFunction on the input stream, thereby creating a transformed output stream. The function will be called for every element in the input streams and can produce zero or more output elements. Contrary to the DataStream.flatMap(FlatMapFunction) function, this function can also query the time and set timers. When reacting to the firing of set timers the function can directly emit elements and/or register yet more timers. Overrides: process(...) in DataStream Type Parameters: The type of elements emitted by the ProcessFunction. Parameters: processFunction The ProcessFunction that is called for each element in the stream. outputType TypeInformation for the result type of the function. Returns: The transformed DataStream. According to error messages, KeyedStream.process need two parameters, the ProcessFunction and R. I try to insert the R parameter but I have no idea what is wrong. Kindly inform me KeyedStream.process function examples including R parameter. Best regards.


Источник: https://stackoverflow.com/questions/780 ... ion-errors