确定哪个对象在 Apache-Flink 中不可序列化

Question

我正在编写一个 Flink 转换器，我有一个具有以下属性的自定义对象Histogram：

case class Histogram(
nRows: Int,
nCols: Int,
min: Int,
step: Double,
private val countMatrix: Array[ArrayBuffer[Double]],
private val cutMatrixL1: Array[ArrayBuffer[Double]],
val distribMatrixL1: Array[ArrayBuffer[Map[Int, Double]]],
private val distribMatrixL2: Array[ArrayBuffer[Map[Int, Double]]],
private val cutMatrixL2: ArrayBuffer[ArrayBuffer[Double]])
extends Serializable {
???
}

这是我FitOperation：

implicit val fitOp = new FitOperation[PIDiscretizerTransformer, LabeledVector] {
override def fit(
instance: PIDiscretizerTransformer,
fitParameters: ParameterMap,
input: DataSet[LabeledVector]): Unit = {
// get params...
val metric = input.map { x ⇒
// (instance, histrogram totalCount)
(x, Histogram(nAttrs, l1InitialBins, min, instance.step), 1)
}.reduce { (m1, m2) ⇒
// Update Layer 1
val updatedL1 = updateL1(m1._1, m1._2, instance.step, initialElems, alpha, m1._3)
//         Update Layer 2 if neccesary
val updatedL2 = if (m1._3 % l2updateExamples == 0) {
updateL2(m1._1, updatedL1)
} else updatedL1
(m2._1, updatedL2, m1._3 + 1)
}.map(_._2)
//      instance.metricsOption = Some(metric)
}
}

这很好用，但是如果我取消注释最后一行：instance.metricsOption = Some(metric)我会得到一个java.io.NotSerializableException: org.apache.flink.api.scala.DataSet

如何找到类中的哪个对象Histogram导致问题？据我所知ArrayBuffer是可序列化的，Map 也是可序列化的。虽然我发现了这个问题：

地图不能在 scala 中序列化？

它说.mapValues不可序列化，但我没有在任何地方使用.mapValues。

Answer 1

问题是你指的是MapFunction内部的instance.step。instance属于无法序列化的PIDiscretizerTransformer类型。因此，您需要计算MapFunction之外的步长并将值传递到函数中。然后，您的程序应该是可序列化的。