title: flink parsing JSON in map: InvalidProgramException: Task not serializable
body: <p>I am working a on Flink project and would like to parse the source JSON string data to Json Object.  I am using <a href="https://github.com/FasterXML/jackson-module-scala" rel="nofollow noreferrer">jackson-module-scala</a> for the JSON parsing. However, I encountered some issues with using the JSON parser within Flink APIs (<code>map</code> for example). </p>
  
  <p>Here are some examples of the code, and I cannot understand the reason under the hood why it is behaving like this.</p>
  
  <h1>Situation 1:</h1>
  
  <p>In this case, I am doing what the <a href="https://github.com/FasterXML/jackson-module-scala#usage" rel="nofollow noreferrer">jackson-module-scala's official exmaple code told me to do</a>:</p>
  
  <ol>
  <li>create a new <code>ObjectMapper</code></li>
  <li>register the <code>DefaultScalaModule</code>
  
  <blockquote>
    <p><em><code>DefaultScalaModule</code> is a Scala object that includes support for all currently supported Scala data types.</em></p>
  </blockquote></li>
  <li>call the <code>readValue</code> in order to parse the JSON to <code>Map</code></li>
  </ol>
  
  <p>The error I got is: <code>org.apache.flink.api.common.InvalidProgramException:</code><strong><code>Task not serializable</code></strong>.</p>
  
  <pre><code>object JsonProcessing {
    def main(args: Array[String]) {
  
      // set up the execution environment
      val env = StreamExecutionEnvironment.getExecutionEnvironment
  
      // get input data
      val text = env.readTextFile("xxx")
  
      val mapper = new ObjectMapper
      mapper.registerModule(DefaultScalaModule)
      val counts = text.map(mapper.readValue(_, classOf[Map[String, String]]))
  
      // execute and print result
      counts.print()
  
      env.execute("JsonProcessing")
    }
  
  }
  </code></pre>
  
  <h1>Situation 2:</h1>
  
  <p>Then I did some Google, and came up with the following solution, where <code>registerModule</code> is moved into the <code>map</code> function.</p>
  
  <pre><code>val mapper = new ObjectMapper
  val counts = text.map(l =&gt; {
    mapper.registerModule(DefaultScalaModule)
    mapper.readValue(l, classOf[Map[String, String]])
  })
  </code></pre>
  
  <p>However, what I am not able to understand is: <strong>why this is going to work, with calling method of outside-defined object <code>mapper</code>?</strong> Is it because the <code>ObjectMapper</code> itself is Serializable as stated here <a href="https://github.com/FasterXML/jackson-databind/blob/master/src/main/java/com/fasterxml/jackson/databind/ObjectMapper.java#L114" rel="nofollow noreferrer">ObjectMapper.java#L114</a>?</p>
  
  <p>Now, the JSON parsing is working fine, but every time, I have to call <code>mapper.registerModule(DefaultScalaModule)</code> which I think could cause some performance issue (<strong><em>Does it?</em></strong>). I also tried another solution as follows.</p>
  
  <h1>Situation 3:</h1>
  
  <p>I created a new <code>case class Jsen</code>, and use it as the corresponding parsing class, registering the Scala modules. And it is also working fine. </p>
  
  <p>However, this is not so flexible if your input JSON is varying very often. It is not maintainable to manage the class <code>Jsen</code>.</p>
  
  <pre><code>case class Jsen(
    @JsonProperty("a") a: String,
    @JsonProperty("c") c: String,
    @JsonProperty("e") e: String
  )
  
  object JsonProcessing {
    def main(args: Array[String]) {
      ...
      val mapper = new ObjectMapper
      val counts = text.map(mapper.readValue(_, classOf[Jsen]))
      ...
  
  }
  </code></pre>
  
  <p>Additionally, I also tried using <code>JsonNode</code> without calling <code>registerModule</code> as follows: </p>
  
  <pre><code>    ...
      val mapper = new ObjectMapper
      val counts = text.map(mapper.readValue(_, classOf[JsonNode]))
      ...
  </code></pre>
  
  <p>It is working fine as well.</p>
  
  <p>My main question is: <strong>what is actually causing the problem of <em>Task not serializable</em> under the hood of <code>registerModule(DefaultScalaModule)</code>?</strong></p>
  
  <p>How to identify whether your code could potentially cause this unserializable problem during coding?</p>