title: java.io.NotSerializableException when save rdd to Hbase using Spark Streaming
body: <p>java.io.NotSerializableException troubles me a lot when i process data with spark.</p>
  
  <pre><code>val hbase_conf = HBaseConfiguration.create()
  hbase_conf.set("hbase.zookeeper.property.clientPort", "2181")
  hbase_conf.set("hbase.zookeeper.quorum", "hadoop-zk0.s.qima-inc.com,hadoop-zk1.s.qima-inc.com,hadoop-zk2.s.qima-inc.com")
  val newAPIJobConfiguration = Job.getInstance(hbase_conf);
  newAPIJobConfiguration.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "mytest_table");
  newAPIJobConfiguration.setOutputFormatClass(classOf[org.apache.hadoop.hbase.mapreduce.TableOutputFormat[ImmutableBytesWritable]])
  newAPIJobConfiguration.getConfiguration().set("mapreduce.output.fileoutputformat.outputdir", "/tmp")
  mydata.foreachRDD( rdd =&gt; {
    val json_rdd = rdd.map(Json.parse _ ).map(_.validate[Scan])
      .map(Scan.transformScanRestult _)
      .filter(_.nonEmpty)
      .map(_.get)
      .map(Scan.convertForHbase _ )
    json_rdd.saveAsNewAPIHadoopDataset(newAPIJobConfiguration.getConfiguration)
  })
  </code></pre>
  
  <p>However it fails cause of java.io.NotSerializableException and follow is error info</p>
  
  <pre><code>17/10/16 18:56:50 ERROR Utils: Exception encountered
          java.io.NotSerializableException: org.apache.hadoop.mapreduce.Job
          at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
          at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
          at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
          at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
          at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
          at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
  </code></pre>
  
  <p>So i change  my code as follows</p>
  
  <pre><code>object mytest_config{
      val hbase_conf = HBaseConfiguration.create()
      hbase_conf.set("hbase.zookeeper.property.clientPort", "2181")
      hbase_conf.set("hbase.zookeeper.quorum", "zk1,zk2")
      val newAPIJobConfiguration = Job.getInstance(hbase_conf);
      newAPIJobConfiguration.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "mytest_table");
      newAPIJobConfiguration.setOutputFormatClass(classOf[org.apache.hadoop.hbase.mapreduce.TableOutputFormat[ImmutableBytesWritable]])
      newAPIJobConfiguration.getConfiguration().set("mapreduce.output.fileoutputformat.outputdir", "/tmp")
    }
  
  mydata.foreachRDD( rdd =&gt; {
        val json_rdd = rdd.map(Json.parse _ )
          .map(_.validate[Scan])
          .map(Scan.transformScanRestult _)
          .filter(_.nonEmpty)
          .map(_.get)
          .map(Scan.convertForHbase _ )
  
       json_rdd.saveAsNewAPIHadoopDataset(mytest_config.newAPIJobConfiguration.getConfiguration)
  })
  </code></pre>
  
  <p>And this could work!
  Somebody got any ideas why this work , and what is the recommended way officially?</p>