了解 Flink - 任务不可序列化

Question

我正在做一个 Flink 项目，遇到了一个问题，我设法在 Stackoverflow 的答案的帮助下解决了这个问题。但是，我不清楚为什么提出的解决方案实际上有效，我发现有关该主题的信息很少。请考虑以下代码：

object DeCP {
def main(args: Array[String]): Unit = {
val params: ParameterTool = ParameterTool.fromArgs(args)
// Get the execution environment and read the data
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val queryPoints: DataSet[Point] = readQueryPoints(env, params)
val points: DataSet[Point] = readFeatureVector(env, params)
// Process the query points
queryPoints
.map(new KNNRich)
.withBroadcastSet(points, "pointsIn")
.print
}
final class KNNRich extends RichMapFunction[Point, (Point, Vector[Point])]{
private var pointsIn: Traversable[Point] = _
override def open(parameters: Configuration): Unit =
pointsIn = getRuntimeContext.getBroadcastVariable[Point]("pointsIn").asScala
def map(queryPoint: Point): (Point, Vector[Point]) = {
val dataSetIn = ExecutionEnvironment.getExecutionEnvironment
.fromCollection(pointsIn.toVector)
val cluster = new Cluster(dataSetIn, queryPoint)
val knn = cluster.kNearestNeighbor(queryPoint, 3) // This call causes problems
(queryPoint, knn.collect.toVector)
}
}
}

群集类和伴随对象定义为：

class Cluster(var points: DataSet[Point],
var clusterLeader: Point) extends Serializable {
private var queryPoint: Point = _
def distance(p: Point): Point = {
p.eucDist(queryPoint)
}
def kNearestNeighbor(queryPoint: Point, k: Int): DataSet[Point] = {
this.queryPoint = queryPoint
this.points.map{p => distance(p)} // Task not serializable
this.points.map{p => p.eucDist(queryPoint)} // Works
this.points.map{p => Cluster.staticDistance(queryPoint, p)} // Works
}
}
object Cluster {
def staticDistance(queryPoint: Point, p: Point): Point = {
p.eucDist(queryPoint)
}
}

对distance方法的调用会导致任务不可序列化异常，但将方法调用替换为定义可以解决此问题。同样，将完全相同的方法定义为伴随对象的成员允许代码正常运行。

为什么第一个调用不起作用，而其他两个调用有效？如果类上有一个更复杂的执行流，并且不容易被替换为伴随对象上的方法，会发生什么情况？

Answer 1

通过执行数据集转换，您只需创建管道的逻辑计划。管道通过调用execute/print/collect提交到集群。

将管道提交到集群时，每个函数(如RichMapFunction(都会序列化、发送到集群、为每个并行实例复制并独立执行。当您收到"任务不可序列化"异常时，这意味着您的RichMapFunction正在传递引用此类外部的变量/对象。您应该确保函数是一个独立的块。

通过调用points.map{}，您将隐式创建一个MapFunction。但是这个MapFunction引用了Cluster的实例，因此不是独立的。Flink 也尝试序列化Cluster但失败了。如果distance是静态的(在伴随对象中定义(，则Cluster也不需要序列化。

顺便说一下，您的示例的另一个问题是您没有按预期使用数据集 API。通常，不应在正在运行的管道中创建管道。这也可能导致意想不到的副作用。