根据Scala-flink中的另一个数据集过滤数据集

我想出了一个解决办法,也许不是最好的,但至少它起作用了。

现在,不是过去 x y 分开的 DataSets ,我路过一个 DataSet[LabeledVector] 只有一列:

val xy = input.map(lv ⇒ LabeledVector(lv.label, DenseVector(lv.vector(0))))

xy 我的职责是:

def conditionalEntropy(xy: DataSet[LabeledVector]): Double = {
    // Get the label
    val y = xy map (_.label)
    // Get probs for the label
    val p = probs(y).toArray.asBreeze
    // Get unique values in label
    val values = y.distinct.collect
    // Compute Conditional Entropy
    val condH = for (i ← values)
      yield entropy(xy.filter(_.label == i))
    p.dot(seq2Breeze(condH))
  }