I am having trouble with resolving this error in Flink (Version 1.11.0): This is how I create a keyed windowed stream: And here is how my ComputeFeatures function looks like: The StreamElement::getId returns a Long so everything regarding types should be correct, but it seems that Flink still has trouble infe…
Tag: flink-streaming
How to see restored value from savepoints?
I have tried to log info about states that recovered when I start a Flink app using savepoints. I see that the app was started using a savepoint correctly. But I can’t log state values that were recovered. The app calculates how many products were bought. Can I use CheckpointedFunction in such way? I se…
Flink error with ‘The implementation of the StreamExecutionEnvironment is not serializable’
I am a beginner of Flink, and I tried to use Flink to run LFM, one of the recommended algorithms, but the following errors appeared in my code when it was running. I tried to find and modify them, but they were not solved. Could someone tell me why I had problems? Here are my main exception And my code
What does latency metrics represent in Flink and are them really valid to evaluate the latency of an application?
I’ve a pipeline developed as shown below: I’m thinking hard about a way to value latency from input to output, but the relation between input and output isn’t 1 to 1 but there are a lot of transformations that make the latency evaluation conceptually very hard. I know about the latency metri…
Filtering duplicates out of an infinite DataStream with windows
I want to filter out duplicates in Flink from an infinite DataStream. I know the duplicates arise only in a small time window (max 10 seconds). I found a promising approach that is pretty simple here. But it doesn’t work. It uses a keyed DataStream and returns only the first message of every window. Thi…
javadoc of SingleOutputStreamOperator#returns(TypeHint typeHint) method
I am reading the source code of SingleOutputStreamOperator#returns, its javadoc is: It mentions FunctionWithNonInferrableReturnType to show case the necessity of returns method, but I am unable to write such a class that is NonInferrableReturnType. Could you please help write a simple one? Thanks! Answer When…
Object reuse – mutating same object – in Flink operators
I was reading the doc here, which gives a use case to reuse the object as given below: So every time instead of creating a new Tuple, it seems to be able to use the same Tuple by using its mutable nature in order to decrease the pressure on GC. Would it be applicable in all operators, where we can
Flink ElasticsearchSinkFunction not serializable in non-static method, but serializable in static method
I have a piece of code that only works inside static methods. If I put the code in a static method, then call it from a non-static method, it works. Never heard of anything like this and couldn’t find information online on it. This works: This doesn’t work: (Full) stack trace: The implementation o…
Compare 2 Data streams in Flink to retrieve missing data
I have 2 Data streams “dataStream1” and “dataStream2”, both of them of type String, i have created 2 tables out of them I need to get all the data in T1 that don’t exist in T2 in time frame of 5 seconds, like if i have inserted a record in T1, it should exist in T2 during 5 secon…
How do I time the checkpointing in Apache Flink streaming?
I am running the Fraud Detector example of Apache Flink with RocksDB as my state backend. I want to know how long does Apache Flink takes to checkpoint the state. My approach is to print time before and after the checkpoint functions. I could not find the function/class or any piece of code that checkpoints t…