Apache Flink-预测处理

Question

我目前正在使用Apache Flink的SVM类来预测一些文本数据。

该类提供了一个预测函数，该函数将DataSet[Vector]作为输入，并作为结果给我一个DataSet[Prediction]。到目前为止还不错。

我的问题是，我没有预测属于哪个文本的上下文，也不能将文本插入到predict((-函数中，然后再进行预测。

代码：

val tweets: DataSet[(SparseVector, String)] =
source.flatMap(new SelectEnglishTweetWithCreatedAtFlatMapper)
.map(tweet => (featureVectorService.transform(tweet._2))
model.predict(tweets).print

result example:
(SparseVector((462,8.73165920153676), (10844,8.508515650222549), (15656,2.931052542245018)),-1.0)

有没有办法把其他数据放在预测旁边，把所有的数据都放在一起？因为没有上下文，预测对我没有帮助。

或者，也许有一种方法可以只预测一个向量而不是一个数据集，我可以在上面的map函数中调用函数。

Answer 1

SVM预测器期望输入Vector的子类型。因此，有两种选择来解决这个问题：

1. 创建Vector的子类型，其中包含作为标记的tweet文本。然后，它将通过预测器循环。这种方法的优点是不需要额外的操作。然而，需要定义新的类和实用程序来用标签表示不同的矢量类型：

val env = ExecutionEnvironment.getExecutionEnvironment
val input = env.fromElements("foobar", "barfo", "test")
val vectorizedInput = input.map(word => {
val value = word.chars().sum()
new DenseVectorWithTag(Array(value), word)
})
val svm = SVM().setBlocks(env.getParallelism)
val weights = env.fromElements(DenseVector(1.0))
svm.weightsOption = Option(weights) // skipping the training here
val predictionResult: DataSet[(DenseVectorWithTag, Double)] = svm.predict(vectorizedInput)
class DenseVectorWithTag(override val data: Array[Double], tag: String)
extends DenseVector(data) {
override def toString: String = "(" + super.toString + ", " + tag + ")"
}

2. 在tweets的矢量化表示上将预测DataSet与输入DataSet连接。这种方法的优点是我们不需要引入新的类。我们为此付出的代价是额外的加入操作，可能会很昂贵：

val input = env.fromElements("foobar", "barfo", "test")
val vectorizedInput = input.map(word => {
val value = word.chars().sum()
(DenseVector(value), word)
})
val svm = SVM().setBlocks(env.getParallelism)
val weights = env.fromElements(DenseVector(1.0))
svm.weightsOption = Option(weights) // skipping the training here
val predictionResult = svm.predict(vectorizedInput.map(a => a._1))
val inputWithPrediction: DataSet[(String, Double)] = vectorizedInput
.join(predictionResult)
.where(0)
.equalTo(0)
.apply((t, p) => (t._2, p._2))