scala - 在 Flink 中调试自定义 Pipeline Transformer
问题描述
我正在尝试按照文档中的指示在 Flink 中实现自定义 Transformer,但是当我尝试执行时,似乎fit
从未调用过该操作。这是我到目前为止所做的:
class InfoGainTransformer extends Transformer[InfoGainTransformer] {
import InfoGainTransformer._
private[this] var counts: Option[collection.immutable.Vector[Map[Key, Double]]] = None
// here setters for params, as Flink does
}
object InfoGainTransformer {
// ====================================== Parameters =============================================
// ...
// ==================================== Factory methods ==========================================
// ...
// ========================================== Operations =========================================
implicit def fitLabeledVectorInfoGain = new FitOperation[InfoGainTransformer, LabeledVector] {
override def fit(instance: InfoGainTransformer, fitParameters: ParameterMap, input: DataSet[LabeledVector]): Unit = {
val counts = collection.immutable.Vector[Map[Key, Double]]()
input.map {
v =>
v.vector.map {
case (i, value) =>
println("INSIDE!!!")
val key = Key(value, v.label)
val cval = counts(i).getOrElse(key, .0)
counts(i) + (key -> cval)
}
}
}
}
implicit def fitVectorInfoGain[T <: Vector] = new FitOperation[InfoGainTransformer, T] {
override def fit(instance: InfoGainTransformer, fitParameters: ParameterMap, input: DataSet[T]): Unit = {
input
}
}
implicit def transformLabeledVectorsInfoGain = {
new TransformDataSetOperation[InfoGainTransformer, LabeledVector, LabeledVector] {
override def transformDataSet(
instance: InfoGainTransformer,
transformParameters: ParameterMap,
input: DataSet[LabeledVector]): DataSet[LabeledVector] = input
}
}
implicit def transformVectorsInfoGain[T <: Vector : BreezeVectorConverter : TypeInformation : ClassTag] = {
new TransformDataSetOperation[InfoGainTransformer, T, T] {
override def transformDataSet(instance: InfoGainTransformer, transformParameters: ParameterMap, input: DataSet[T]): DataSet[T] = input
}
}
}
然后我尝试以两种方式使用它:
val scaler = StandardScaler()
val polyFeatures = PolynomialFeatures()
val mlr = MultipleLinearRegression()
val gain = InfoGainTransformer().setK(2)
// Construct the pipeline
val pipeline = scaler
.chainTransformer(polyFeatures)
.chainTransformer(gain)
.chainPredictor(mlr)
val r = pipeline.predict(dataSet map (_.vector))
r.print()
只有我的变压器:
pipeline.fit(dataSet)
在这两种情况下,当我在内部设置断点时fitLabeledVectorInfoGain
,例如在 lineinput.map
中,调试器会停在那里,但如果我还在嵌套映射中设置断点,例如 bellow println("INSIDE!!!")
,它永远不会停在那里。
有谁知道我该如何调试这个自定义变压器?
解决方案
看来它现在正在工作。我认为发生的事情是我没有正确实施,FitOperation
因为实例状态中没有保存任何内容,这就是现在的实施:
implicit def fitLabeledVectorInfoGain = new FitOperation[InfoGainTransformer, LabeledVector] {
override def fit(instance: InfoGainTransformer, fitParameters: ParameterMap, input: DataSet[LabeledVector]): Unit = {
// val counts = collection.immutable.Vector[Map[Key, Double]]()
val r = input.map {
v =>
v.vector.foldLeft(Map.empty[Key, Double]) {
case (m, (i, value)) =>
println("INSIDE fit!!!")
val key = Key(value, v.label)
val cval = m.getOrElse(key, .0) + 1.0
m + (key -> cval)
}
}
instance.counts = Some(r)
}
}
现在调试器正确进入所有断点并且TransformOperation
它也被调用。
推荐阅读
- javascript - 与 mlab 的连接不断失败 - 第一个真实世界的应用程序-
- python - tkinter 如何检测窗口是否降低(.lower() 方法)
- c - 尝试使用简单的计时器和按位 & 检查为精灵设置动画
- linux - Ubuntu 20.04 上的 PhpStorm 后退导航中断
- swift - 我想改变 tableview 的 rowHeight,同时改变方向它在模拟器中工作而不是在真实设备中
- fastapi - FastAPI graphql 订阅
- r - 如何在使用 tidymodels 时取消缩放和理解 glmnet 系数?
- mongodb - MongoDB Spring 聚合排除不起作用
- powershell - PowerShell Set-Partition -NewDriveLetter 返回 Set-Partition:请求的访问路径已在使用中。活动 ID {
- javascript - 二进制插入,数组的后半部分没有被递归调用?