I'm using Flink DataStreams to join 2 streams (a Book stream and a Publisher stream). I'm trying to remove elements by using evictor in case they are deleted from the database, which is indicated with the variable deleted.
When I run the code without the evictor it works well, but when I add the evictor it fails.
DataStream<BooksWithPublishers> book_publisher = bookStream
.join(publishStream)
.where(value -> value.publisherId)
.equalTo(value -> value.id)
.window(GlobalWindows.create())
.trigger(new ForeverTrigger<>())
.evictor(new Evictor<CoGroupedStreams.TaggedUnion<Book, Publisher>, GlobalWindow>() {
@Override
public void evictBefore(Iterable<TimestampedValue<CoGroupedStreams.TaggedUnion<Book, Publisher>>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
Iterator<TimestampedValue<CoGroupedStreams.TaggedUnion<Book, Publisher>>> it = elements.iterator();
while(it.hasNext()){
CoGroupedStreams.TaggedUnion<Book, Publisher> cg = it.next().getValue();
Book book = cg.getOne();
Publisher pub = cg.getTwo();
if(book.deleted || pub.deleted){
it.remove();
}
}
}
@Override
public void evictAfter(Iterable<TimestampedValue<CoGroupedStreams.TaggedUnion<Book, Publisher>>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
}
})
.apply(new JoinFunction<Book, Publisher, BooksWithPublishers>() {
@Override
public BooksWithPublishers join(Book first, Publisher second) throws Exception {
return new BooksWithPublishers(first.id,first.releaseDate,first.title,second);
}
});
Error:
org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot serialize operator object class org.apache.flink.streaming.api.operators.SimpleUdfStreamOperatorFactory.
at org.apache.flink.streaming.api.graph.StreamConfig.setStreamOperatorFactory(StreamConfig.java:304)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.setVertexConfig(StreamingJobGraphGenerator.java:694)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:438)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:399)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:390)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.setChaining(StreamingJobGraphGenerator.java:356)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:179)
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:116)
at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:908)
at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:50)
at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:39)
at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:56)
at org.apache.flink.client.deployment.executors.LocalExecutor.getJobGraph(LocalExecutor.java:104)
at org.apache.flink.client.deployment.executors.LocalExecutor.execute(LocalExecutor.java:82)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1905)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1796)
at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:69)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1782)
at com.x.x.pipelines.Bootstrapper.boot(Bootstrapper.java:68)
at com.x.x.Main.main(Main.java:7)
Caused by: java.io.NotSerializableException: com.x.x.pipelines.library.author.index.AuthorIndex
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:586)
at org.apache.flink.util.InstantiationUtil.writeObjectToConfig(InstantiationUtil.java:515)
at org.apache.flink.streaming.api.graph.StreamConfig.setStreamOperatorFactory(StreamConfig.java:301)
... 19 more
I've tried running the code with .evictor() with empty methods and it still gave me an error.
Why can't I use evictor()?
The problem is most likely that your enclosing class (AuthorIndex presumably) is not serializable and your program is trying to serialize it. This can be avoided by creating a separate class instead of using an anonymous class or making the method static.