我有一个 DataSet
从csv文件读取:
val dataSet = env.readCsvFile[ElecNormNew](
getClass.getResource("/elecNormNew.arff").getPath,
pojoFields = Array("date", "day", "period", "nswprice", "nswdemand", "vicprice", "vicdemand", "transfer", "label")
据我所知, ElecNormNew
是pojo:
// elecNormNew POJO
class ElecNormNew(
var date: Double,
var day: Int,
var period: Double,
var nswprice: Double,
var nswdemand: Double,
var vicprice: Double,
var vicdemand: Double,
var transfer: Double,
var label: String) extends Serializable {
def this() = {
this(0, 0, 0, 0, 0, 0, 0, 0, "")
}
}
我还有一个简单的课程:
case class Discretizer[T](
data: DataSet[T],
nBins: Int = 5,
s: Int = 1000) {
private[this] val log = LoggerFactory.getLogger("Discretizer")
private[this] val V = Vector.tabulate(10)(_ => IntervalHeap(nBins, 1, 1, s))
private[this] def updateSamples(x: T): Vector[IntervalHeap] = {
log.warn(s"$x")
V
}
def discretize() = {
data map (x => updateSamples(x))
}
}
但是当我试着使用它时,例如从一个测试中:
val a = new Discretizer[ElecNormNew](dataSet)
a.discretize
我得到以下错误:
org.apache.flink.api.common.InvalidProgramException: Task not serializable
// ...
[info] at com.elbauldelprogramador.discretizers.IDADiscretizer.discretize(IDADiscretizer.scala:69)
// ...
[info] Cause: java.io.NotSerializableException: org.apache.flink.api.scala.DataSet
// ...
我读过这些问题和答案,但运气不好:
任务不可序列化flink
任务在scala中不可序列化
任务不可序列化:java.io.notserializableexception仅在类而不是对象上调用闭包外的函数时发生异常
1条答案
按热度按时间z9zf31ra1#
我想说你提到的第一个链接提供了答案:
问题是您从mapfunction中引用数据集页。这是不可能的,因为数据集只是数据流的逻辑表示,不能在运行时访问。
discretize
使用map
所以这也适用于这里。