title: How to join two Hbase tables in Spark using Scala
body: <p>I have two tables in HBase that I need to join using scala. The tables are imported from Oracle using sqoop and are available for querying in the Hue data browser </p>
  
  <p>Using Spark 1.5, Scala 2.10.4.</p>
  
  <p>I'm using the HBase data connector from here: <a href="https://github.com/nerdammer/spark-hbase-connector" rel="nofollow noreferrer">https://github.com/nerdammer/spark-hbase-connector</a></p>
  
  <pre><code>import it.nerdammer.spark.hbase._
  import org.apache.hadoop.hbase.client.{ HBaseAdmin, Result }
  import org.apache.hadoop.hbase.{ HBaseConfiguration, HTableDescriptor }
  import org.apache.hadoop.hbase.mapreduce.TableInputFormat
  import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  import org.apache.spark._
  import it.nerdammer.spark.hbase.conversion.{ FieldReader, FieldWriter }
  import org.apache.hadoop.hbase.util.Bytes
  
  case class Artist(id: String,
                   name: String,
                   age: Int);
  
  case class Cd(id: String,
                artistId: String,
                title: String,
                year: Int);
  
  case class ArtistCd(id: String,
                      name: String,
                      title: String,
                      year: Int);
  
  implicit def artistReader: FieldReader[Artist] = new FieldReader[Artist] {
  
      override def map(data: HBaseData): Artist = Artist(
  
          id = Bytes.toString(data.head.get),
          name = Bytes.toString(data.drop(1).head.get),
          age = Bytes.toInt(data.drop(2).head.get));
  
      override def columns = Seq("NAME", "AGE");
  
  };
  
  implicit def cdReader: FieldReader[Cd] = new FieldReader[Cd] {
  
      override def map(data: HBaseData): Cd = Cd(
  
          id = Bytes.toString(data.head.get),
          artistId = Bytes.toString(data.drop(1).head.get),
          title = Bytes.toString(data.drop(2).head.get),
          year = Bytes.toInt(data.drop(3).head.get));
  
      override def columns = Seq("ARTIST_ID", "TITLE", "YEAR");
  
  };
  
  implicit def artistCdWriter: FieldWriter[ArtistCd] = new FieldWriter[ArtistCd] {
      override def map(data: ArtistCd): HBaseData =
          Seq(
              Some(Bytes.toBytes(data.id)),
              Some(Bytes.toBytes(data.name)),
              Some(Bytes.toBytes(data.title)),
              Some(Bytes.toBytes(data.year)));
  
      override def columns = Seq("NAME", "TITLE", "YEAR");
  };
  
  val conf = new SparkConf().setAppName("HBase Join").setMaster("spark://localhost:7337")
  val sc = new SparkContext(conf)
  
  val artistRDD = sc.hbaseTable[Artist]("ARTISTS").inColumnFamily("cf")
  val cdRDD = sc.hbaseTable[Cd]("CDS").inColumnFamily("cf")
  
  val artistById = artistRDD.keyBy(f =&gt; f.id)
  val cdById = cdRDD.keyBy(f =&gt; f.artistId)
  
  val artistcd = artistById.join(cdById)
  
  val artistCdRDD = artistcd.map(f =&gt; new ArtistCd(f._2._1.id, f._2._2.title, f._2._1.name, f._2._2.year))
  artistCdRDD.toHBaseTable("ARTIST_CD").inColumnFamily("cf").save()
  System.exit(1)
  </code></pre>
  
  <p>When I run this I get the following exception </p>
  
  <pre><code>16/01/22 14:27:04 WARN TaskSetManager: Lost task 0.0 in stage 5.0 (TID 3, localhost): org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException: Failed 2068 actions: ARTIST_CD: 2068 times,
          at org.apache.hadoop.hbase.client.AsyncProcess$BatchErrors.makeException(AsyncProcess.java:227)
          at org.apache.hadoop.hbase.client.AsyncProcess$BatchErrors.access$1700(AsyncProcess.java:207)
          at org.apache.hadoop.hbase.client.AsyncProcess.waitForAllPreviousOpsAndReset(AsyncProcess.java:1663)
          at org.apache.hadoop.hbase.client.BufferedMutatorImpl.backgroundFlushCommits(BufferedMutatorImpl.java:208)
          at org.apache.hadoop.hbase.client.BufferedMutatorImpl.doMutate(BufferedMutatorImpl.java:141)
          at org.apache.hadoop.hbase.client.BufferedMutatorImpl.mutate(BufferedMutatorImpl.java:98)
          at org.apache.hadoop.hbase.mapreduce.TableOutputFormat$TableRecordWriter.write(TableOutputFormat.java:129)
          at org.apache.hadoop.hbase.mapreduce.TableOutputFormat$TableRecordWriter.write(TableOutputFormat.java:85)
          at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply$mcV$sp(PairRDDFunctions.scala:1036)
          at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1034)
          at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1034)
          at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1206)
          at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1042)
          at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1014)
          at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
          at org.apache.spark.scheduler.Task.run(Task.scala:88)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
          at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>If anyone has any experience in this, I'd really appreciate your help</p>
  
  <p>I've seen the two solutions here <a href="https://stackoverflow.com/questions/25589288/how-to-join-two-tables-in-hbase">How to Join two tables in Hbase</a> and <a href="https://stackoverflow.com/questions/11327316/how-to-join-tables-in-hbase">how to join tables in hbase</a> and unfortunately neither is going to work for me</p>