title: Task not serializable in Flink even with POJO
body: <p>I have a <code>DataSet</code> read from a CSV file:</p>
  
  <pre><code>val dataSet = env.readCsvFile[ElecNormNew](
        getClass.getResource("/elecNormNew.arff").getPath,
        pojoFields = Array("date", "day", "period", "nswprice", "nswdemand", "vicprice", "vicdemand", "transfer", "label")
  </code></pre>
  
  <p>As far as I know, <code>ElecNormNew</code> is a POJO:</p>
  
  <pre><code>// 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, "")
    }
  }
  </code></pre>
  
  <p>I also have a simple class:</p>
  
  <pre><code>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)(_ =&gt; IntervalHeap(nBins, 1, 1, s))
  
    private[this] def updateSamples(x: T): Vector[IntervalHeap] = {
      log.warn(s"$x")
      V
    }
  
    def discretize() = {
      data map (x =&gt; updateSamples(x))
    }
  }
  </code></pre>
  
  <p>But when I try to use it, from a test for example:</p>
  
  <pre><code>val a = new Discretizer[ElecNormNew](dataSet)
  a.discretize
  </code></pre>
  
  <p>I am getting the following error:</p>
  
  <pre><code>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
  // ...
  </code></pre>
  
  <p>I've read this questions and its answers, without luck:</p>
  
  <ul>
  <li><a href="https://stackoverflow.com/questions/31315478/task-not-serializable-flink">Task not serializable Flink</a></li>
  <li><a href="https://stackoverflow.com/questions/43910460/task-not-serializable-in-scala">Task not serializable in scala</a></li>
  <li><a href="https://stackoverflow.com/questions/22592811/task-not-serializable-java-io-notserializableexception-when-calling-function-ou">Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects</a></li>
  </ul>