Flink with Guava cache - ProcessFunction 的实现不可序列化答案

【问题标题】：Flink with Guava cache - The implementation of the ProcessFunction is not serializableFlink with Guava cache - ProcessFunction 的实现不可序列化
【发布时间】：2019-08-23 08:58:18
【问题描述】：

我实现了一个 ProcessFunction，它使用 Guava 缓存来过滤传入事件流。代码如下所示：

object myJob {
 private def updateCache(cacheObject, someValue) = {}
 private def getCacheValue(cacheObject, someKey) = {}

 override def run(params, executionEnv) = {
  val inputStream = executionEnv.stream

  val c = CacheBuilder.newBuilder()

  val outStream = inputStream.process(new ProcessFunction() { 
    updateCache()
    getCacheValue} 
    )
 }
}

在向 Flink 提交作业时，我收到以下错误：

Caused by: org.apache.flink.api.common.InvalidProgramException: The implementation of the ProcessFunction is not serializable. The object probably contains or references non serializable fields.
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:99)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1560)
at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:185)
at org.apache.flink.streaming.api.datastream.DataStream.process(DataStream.java:666)
at org.apache.flink.streaming.api.scala.DataStream.process(DataStream.scala:686)

知道我做错了什么吗？如何解决此序列化错误？

【问题讨论】：

我通过将缓存对象声明移到运行函数之外解决了这个问题。将其设为私有值。

标签： scala serialization guava apache-flink flink-streaming

【解决方案1】：

该错误基本上表明您依赖于 Flink 不可序列化的 Object。在您展示的情况下，使用加载器将字段标记为惰性应该可以解决问题：

   lazy val c = CacheBuilder.newBuilder()

一般情况下，你应该参考Flink的documentation，它解释了这个问题

【讨论】：