java - Flink 数据流驱逐者
问题描述
我正在使用 Flink DataStreams 加入 2 个流(一个 Book 流和一个 Publisher 流)。我试图通过使用删除元素evictor
,以防它们从数据库中删除,这用删除的变量表示。
当我在没有evictor
它的情况下运行代码时效果很好,但是当我添加evictor
它时它会失败。
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);
}
});
错误:
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
我尝试使用 .evictor() 和空方法运行代码,但它仍然给我一个错误。
为什么我不能使用 evictor()?
解决方案
问题很可能是您的封闭类(大概是AuthorIndex)不可序列化,并且您的程序正在尝试对其进行序列化。这可以通过创建单独的类而不是使用匿名类或使方法静态来避免。
推荐阅读
- docker - 使用 JDBC 输入插件在 Docker 上的 Logstash 不会从 SQL Server 获取所有行
- flutter - 如何突出显示我在字符串中搜索的确切单词,而不仅仅是字符串的开头?
- flutter - 如何实时获取文档长度
- javascript - Reactjs 从获取请求中返回一个对象
- javascript - Nginx 正在尝试打开文件而不是重定向到代理
- python - _joint_log_likelihood 给我错误的值
- sql - 如何按日期获取最后一件商品的价格
- python-3.x - 定期执行功能而不停止其他操作
- r - 使用 R 从 PDF 表单到数据框的文本挖掘
- r - 如果 r 中出现错误,则转到 lapply() 的下一次迭代