added Hbase methods

added Hbase methods

Author: gaido

SparkInputLib/build.sbt

@@ -2,11 +2,19 @@ name := "spark_input_lib"
 
 organization := "it.mgaido"
 
-version := "0.0.3"
+version := "0.0.4"
 
 scalaVersion := "2.10.5"
 
-libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.2.0-cdh5.3.2" % "provided"
+autoScalaLibrary := false
+
+libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.1" % "provided"
+
+libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.1.2" % "provided" exclude("javax.servlet", "servlet-api")
+libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.1.2" % "provided" exclude("javax.servlet", "servlet-api")
+libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.1.2" % "provided" exclude("javax.servlet", "servlet-api")
+
+
 libraryDependencies += "org.scalatest" % "scalatest_2.10" % "2.1.3" % "test"
 
 resolvers += "Cloudera Repository" at "https://repository.cloudera.com/artifactory/cloudera-repos/"

SparkInputLib/src/main/scala-2.10/it/mgaido/spark/io/IOHelper.scala

@@ -4,6 +4,13 @@ import org.apache.spark.rdd.RDD
 import org.apache.spark.SparkContext
 import org.apache.hadoop.io.LongWritable
 import org.apache.hadoop.io.Text
+import org.apache.hadoop.hbase.client.Result
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable
+import org.apache.hadoop.hbase.HBaseConfiguration
+import org.apache.hadoop.hbase.mapreduce.TableInputFormat
+import org.apache.hadoop.mapreduce.Job
+import org.apache.hadoop.hbase.client.Put
+import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
 
 /**
  * @author m.gaido
@@ -19,4 +26,26 @@ object IOHelper {
           .map(line => line._2.toString)
   }
 
+  def readHbaseTable(tableName:String):RDD[(ImmutableBytesWritable, Result)] = {
+    @transient val hconf = HBaseConfiguration.create()
+    hconf.set(TableInputFormat.INPUT_TABLE, tableName)
+    val job = Job.getInstance(hconf)
+    job.setInputFormatClass(classOf[TableInputFormat])
+    
+    @transient val jobConf = job.getConfiguration
+    jobConf.set(TableInputFormat.INPUT_TABLE, tableName)
+    SparkContext.getOrCreate().newAPIHadoopRDD(jobConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])
+  }
+  
+  def upsertToHbaseTable(tableName:String, rdd:RDD[(ImmutableBytesWritable, Put)]) = {
+    @transient val hconf = HBaseConfiguration.create()
+    hconf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
+    
+    val job = Job.getInstance(hconf)
+    job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
+    @transient val jobConf = job.getConfiguration
+    jobConf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
+    rdd.saveAsNewAPIHadoopDataset(jobConf)
+  }
+  
 }

SparkInputLib/src/main/scala-2.10/it/mgaido/spark/io/package.scala

@@ -5,14 +5,22 @@ package it.mgaido.spark
  */
 import org.apache.spark.rdd.RDD
 import org.apache.spark.SparkContext
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable
+import org.apache.hadoop.hbase.client.Put
 
 package object io {
   implicit class IOSparkContext(sparkContext: SparkContext){
 
     def textFileWithHeader(path:String, numHeaderLines:Int):RDD[String] = {
       IOHelper.readTextFilesWithHeader(sparkContext,path, numHeaderLines)
     }
+    
+    def hbaseTable(tableName:String) = IOHelper.readHbaseTable(tableName)
 
   }
+  
+  implicit class HBaseRDD(rdd:RDD[(ImmutableBytesWritable, Put)]){
+    def upsertToHbaseTable(tableName:String) = IOHelper.upsertToHbaseTable(tableName, rdd)
+  }
 }