scala—确定在ApacheFlink中哪个对象不可序列化

kuhbmx9i  于 2021-06-21  发布在  Flink
关注(0)|答案(1)|浏览(289)

我正在写一个flink transformer,我有一个自定义对象 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也是。尽管我发现了这个问题:
Map不能在scala中序列化?
上面写着 .mapValues 不可序列化,但我没有使用 .mapValues 任何地方。

3b6akqbq

3b6akqbq1#

问题是你指的是 instance.step 在你的 MapFunction . instance 属于类型 PIDiscretizerTransformer 无法序列化的。因此,您需要在 MapFunction 并将值传递给函数。那么你的程序应该是可序列化的。

相关问题