2

I am trying to learn spark + scala. I want to read from HBase, but without mapreduce. I created a simple HBase table - "test" and did 3 puts in it. I want to read it via spark (without HBaseTest which uses mapreduce). I tried to run the following commands on shell

val numbers = Array(
  new Get(Bytes.toBytes("row1")), 
  new Get(Bytes.toBytes("row2")), 
  new Get(Bytes.toBytes("row3")))
val conf = new HBaseConfiguration()
val table = new HTable(conf, "test")
sc.parallelize(numbers, numbers.length).map(table.get).count()

I keep getting error - org.apache.spark.SparkException: Job aborted: Task not serializable: java.io.NotSerializableException: org.apache.hadoop.hbase.HBaseConfiguration

Can someone help me , how can I create a Htable which uses serialzable configuration

thanks

samthebest
  • 30,803
  • 25
  • 102
  • 142
user3529980
  • 61
  • 2
  • 5

2 Answers2

4

Your problem is that table is not serializable (rather it's member conf) and your trying to serialize it by using it inside a map. They way your trying to read HBase isn't quite correct, it looks like your trying some specific Get's and then trying to do them in parallel. Even if you did get this working, this really wouldn't scale as your going to perform random reads. What you want to do is perform a table scan using Spark, here is a code snippet that should help you do it:

val conf = HBaseConfiguration.create()
conf.set(TableInputFormat.INPUT_TABLE, tableName)

sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
  classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
  classOf[org.apache.hadoop.hbase.client.Result])

This will give you an RDD containing the NaviagableMap's that constitute the rows. Below is how you can change the NaviagbleMap to a normal Scala map of Strings:

...
.map(kv => (kv._1.get(), navMapToMap(kv._2.getMap)))
.map(kv => (Bytes.toString(kv._1), rowToStrMap(kv._2)))

def navMapToMap(navMap: HBaseRow): CFTimeseriesRow =
  navMap.asScala.toMap.map(cf =>
    (cf._1, cf._2.asScala.toMap.map(col =>
      (col._1, col._2.asScala.toMap.map(elem => (elem._1.toLong, elem._2))))))

def rowToStrMap(navMap: CFTimeseriesRow): CFTimeseriesRowStr =
  navMap.map(cf =>
    (Bytes.toString(cf._1), cf._2.map(col =>
      (Bytes.toString(col._1), col._2.map(elem => (elem._1, Bytes.toString(elem._2)))))))

Final point, if you really do want to try to perform random reads in parallel I believe you might be able to put the HBase table initialization inside the map.

samthebest
  • 30,803
  • 25
  • 102
  • 142
  • Hello, Sorry for the delayed response. I want to try this way of reading, can you please help me to define the class in a serializable way. I want to try to distribute the Keys and compare its performance with standard way. Any idea to do this will be deeply appreciated. – user3529980 May 25 '14 at 08:30
2

what happens when you do

@transient val conf = new HBaseConfiguration

UPDATE Apparently there are other parts of the HBase submitted task that are also not serializable. Each of these will need to be addressed.

  • Consider whether the entity will have the same meaning/semantics on both sides of the wire. Any connections will certainly not. The HBaseConfiguration should not be serialized. But primitives and simple objects built atop primitives - and not containing context-sensitive data - are fine to include in the serialization

  • For context-sensitive entities - including the HBaseConfiguration and any connection oriented data structures - you should mark them @transient and then in the readObject() method they should be instantiated with values relevant to the client environment.

WestCoastProjects
  • 58,982
  • 91
  • 316
  • 560
  • Hello, tried that. Now the error is - org.apache.spark.SparkException: Job aborted: Task not serializable: java.io.NotSerializableException: org.apache.hadoop.hbase.client.Get need some mechanism by which I can make the class serializable – user3529980 May 13 '14 at 00:13
  • Hmm, it should work @javadba where did you learn this? I read this http://examples.javacodegeeks.com/java-basics/exceptions/java-io-notserializableexception-how-to-solve-not-serializable-exception/ and it says it should work (also see http://www.scala-lang.org/old/node/106) – samthebest Aug 12 '14 at 17:27
  • 1
    @user3529980 Try adding @transient above the `val table` too and move the `new Get` inside the lambda. `Get` is not serializable either, so you'll need to create the `Get` inside the lamda. – samthebest Aug 12 '14 at 19:43
  • @samthebest Thanks for the additional comments. I have updated my answer for some general pointers. – WestCoastProjects Aug 12 '14 at 20:24