java.io.NotSerializableException when save rdd to Hbase using Spark Streaming

Question

java.io.NotSerializableException troubles me a lot when i process data with spark.

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

However it fails cause of java.io.NotSerializableException and follow is error info

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)

So i change my code as follows

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

And this could work! Somebody got any ideas why this work , and what is the recommended way officially?

mrsrinivas mrsrinivas · Accepted Answer · 2017-10-16T13:40:11

That ERROR is due to

This newAPIJobConfiguration was initialized in the driver

val newAPIJobConfiguration = Job.getInstance(hbase_conf);

It's using inside workers(foreach)

json_rdd.saveAsNewAPIHadoopDataset(newAPIJobConfiguration.getConfiguration)

java.io.NotSerializableException when save rdd to Hbase using Spark Streaming

1 Answers