Flink中函数调用之间的类状态获取丢失

我有这门课:

case class IDADiscretizer(
  nAttrs: Int,
  nBins: Int = 5,
  s: Int = 5) extends Serializable {

  private[this] val log = LoggerFactory.getLogger(this.getClass)
  private[this] val V = Vector.tabulate(nAttrs)(i => new IntervalHeapWrapper(nBins, i))
  private[this] val randomReservoir = SamplingUtils.reservoirSample((1 to s).toList.iterator, 1)

  def updateSamples(v: LabeledVector): Vector[IntervalHeapWrapper] = {
    val attrs = v.vector.map(_._2)
    val label = v.label
    // TODO: Check for missing values
    attrs
      .zipWithIndex
      .foreach {
        case (attr, i) =>
          if (V(i).getNbSamples < s) {
            V(i) insertValue attr // insert
          } else {
            if (randomReservoir(0) <= s / (i + 1)) {
              //val randVal = Random nextInt s
              //V(i) replace (randVal, attr)
              V(i) insertValue attr
            }
          }
      }
    V
  }

  /**
   * Return the cutpoints for the discretization
   *
   */
  def cutPoints: Vector[Vector[Double]] = V map (_.getBoundaries.toVector)

  def discretize(data: DataSet[LabeledVector]): (DataSet[Vector[IntervalHeapWrapper]], Vector[Vector[Double]]) = {
    val r = data map (x => updateSamples(x))
    val c = cutPoints

    (r, c)
  }
}

使用Flink,我想在 discretize ,但似乎存储在 V 得到损失。我必须用吗 Broadcast 就像在 this question ?有没有更好的方法来访问类的状态?

我试过打电话 cutpoints 有两种方式,一种是:

def discretize(data: DataSet[LabeledVector]) = data map (x => updateSamples(x))

然后,从外面呼叫:

val a = IDADiscretizer(nAttrs = 4)
val r = a.discretize(dataSet)
r.print
val cuts = a.cutPoints

这里,切割是空的,所以我试图计算离散化以及内部的切割点。 离散化 :

def discretize(data: DataSet[LabeledVector]) = {
    val r = data map (x => updateSamples(x))
    val c = cutPoints

    (r, c)
  }

像这样使用:

val a = IDADiscretizer(nAttrs = 4)
val (d, c) = a.discretize(dataSet)
c foreach println

但同样的事情发生了。

最后,我也试着 V 完全公开:

val V = Vector.tabulate(nAttrs)(i => new IntervalHeapWrapper(nBins, i))

仍然是空的

我做错什么了?

相关问题:

• Keep keyed state across multiple transformations
• Flink State backend keys atomicy and distribution
• Flink: does state access across stream?
• Flink: Sharing state in CoFlatMapFunction

回答

多亏了@tillrohrmann,我终于做到了:

private[this] def computeCutPoints(x: LabeledVector) = {
    val attrs = x.vector.map(_._2)
    val label = x.label
    attrs
      .zipWithIndex
      .foldLeft(V) {
        case (iv, (v, i)) =>
          iv(i) insertValue v
          iv
      }
  }

  /**
   * Return the cutpoints for the discretization
   *
   */
  def cutPoints(data: DataSet[LabeledVector]): Seq[Seq[Double]] =
    data.map(computeCutPoints _)
      .collect
      .last.map(_.getBoundaries.toVector)

  def discretize(data: DataSet[LabeledVector]): DataSet[LabeledVector] =
data.map(updateSamples _)

然后像这样使用:

val a = IDADiscretizer(nAttrs = 4)
val d = a.discretize(dataSet)
val cuts = a.cutPoints(dataSet)
d.print
cuts foreach println

我不知道这是不是最好的方法,但至少现在在工作。