怎么在Spark Core之上使用hbase-rdd擴建自己的模塊

這篇文章主要為大家展示了“怎么在Spark Core之上使用hbase-rdd擴建自己的模塊”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“怎么在Spark Core之上使用hbase-rdd擴建自己的模塊”這篇文章吧。

hbase-rdd是一個構建在SparkContext基礎之上的用于對Hbase進行增刪改查的第三方開源模塊，目前***版本為0.7.1。目前該rdd在操作hbase時，默認調用隱式方法。

implicitdef stringToBytes(s: String): Array[Byte] = {  Bytes.toBytes(s)  }

將RDD的key轉換成字節b，然后調用Hbase的put(b)方法保存rowkey，之后將RDD的每一行存入hbase。

在軌跡圖繪制項目數據計算中，我們考慮到hbase的rowkey的設計——盡量減少rowkey存儲的開銷。雖然hbase-rdd最終的rowkey默認都是采用字節數組，但這個地方我們希望按自己的方式組裝rowkey。使用MD5(imei)+dateTime組成的字節數組作為rowkey。因此默認的hbase-rdd提供的方法是不滿足我們存儲需求的，需要對源代碼進行修改。在toHbase方法中，有一個convert方法，該方法將對RDD中的每一行數據進行轉化，使用RDD中的key生成Put(Bytes.toBytes(key))對象，該對象為之后存儲Hbase提供rowkey。

在convert函數中，對其實現進行了改造，hbase-rdd默認使用stringToBytes隱式函數將RDD的String類型的key轉換成字節數組，這里我們需要改造，不使stringToBytes隱式方法，而是直接生成字節數據。

protected def convert(id: String, values: Map[String, Map[String, A]], put: PutAdder[A]) = {  val strs = id.split(",")  val imei = strs {0}  val dateTime = strs {1}  val b1 = MD5Utils.computeMD5Hash(imei.getBytes())  val b2 = Bytes.toBytes(dateTime.toLong)  val key = b1.++(b2)  val p = new Put(key)//改造  var empty = true  for {  (family, content) <- values  (key, value) <- content  } {  empty = false  if (StrUtils.isNotEmpty(family) &&StrUtils.isNotEmpty(key)) {  put(p, family, key, value)  }  }  if (empty) None else Some(new ImmutableBytesWritable, p)  }

這樣就實現了使用自己的方式構建rowkey，當然基于此思想我們可以使用任意的方式構建rowkey。

在使用hbase-rdd插件的過程中，我在思考，默認的RDD上是沒有toHbase方法的，那為什么引入hbase-rdd包之后，RDD之上就有toHbase方法了?經過查看源碼，發現hbase-rdd包中提供了兩個隱式方法：

implicitdef toHBaseRDDSimple[A](rdd: RDD[(String, Map[String, A])])(implicit writer: Writes[A]): HBaseWriteRDDSimple[A] =new HBaseWriteRDDSimple(rdd, pa[A]) implicit def toHBaseRDDSimpleTS[A](rdd: RDD[(String, Map[String, (A, Long)])])(implicit writer: Writes[A]): HBaseWriteRDDSimple[(A, Long)] =new HBaseWriteRDDSimple(rdd, pa[A])

這兩個方法在發現RDD上沒有toHbase方法時會自動嘗試調用，從隱式定義中嘗試找到解決方案，嘗試之后發現有定義toHBaseRDDSimple隱式方法，于是調用該隱式方法新建HBaseWriteRDDSimple類，返回hBaseWriteRDDSimple，而在hBaseWriteRDDSimple對象中是有toHbase方法的，因此在引入hbase-rdd之后，可以發現原本沒有toHbase方法的RDD上有toHbase方法了。這一切都要歸功于Scala強大的隱式轉換功能。

那明白了原理，是否我們可以基于RDD寫自己的模塊，說干就干!

***步：新建Trait

traitHaha{ implicitdef gaga[A](rdd: RDD[String]): Hehe= newHehe(rdd) }

第二步：新建Hehe類

final  class Hehe(rdd:RDD[String]) { def wow(tableName:String,family:String): Unit ={ println("---------------------------------------------") println("tableName:"+tableName+" - family:"+family) println("size:"+rdd.count()) rdd.collect().foreach(data=>println(data)) println("---------------------------------------------")    } }

第三步：新建包對象

package object test extends Haha

第四步：新建test類

object Test{ def main(args: Array[String]) { valsparkConf = new SparkConf().setAppName("Test") valsc = new SparkContext(sparkConf) sc.makeRDD(Seq("one","two","three","four")).wow("taskDataPre","T")   } }

項目結構圖：

運行效果圖：

以上是“怎么在Spark Core之上使用hbase-rdd擴建自己的模塊”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！