Task not serializable in Flink even with POJO

Question

I have a DataSet read from a CSV file:

val dataSet = env.readCsvFile[ElecNormNew](
      getClass.getResource("/elecNormNew.arff").getPath,
      pojoFields = Array("date", "day", "period", "nswprice", "nswdemand", "vicprice", "vicdemand", "transfer", "label")

As far as I know, ElecNormNew is a 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, "")
  }
}

I also have a simple class:

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))
  }
}

But when I try to use it, from a test for example:

val a = new Discretizer[ElecNormNew](dataSet)
a.discretize

I am getting the following error:

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
// ...

I've read this questions and its answers, without luck:

• Task not serializable Flink
• Task not serializable in scala
• Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects

Answer 1

I would say the first link you have mentioned provides the answer:

The problem is that you reference the DataSet pages from within a MapFunction. This is not possible, since a DataSet is only the logical representation of a data flow and cannot be accessed at runtime.

discretize uses map, so that applies also here.