建站
咨詢
售后
建站咨詢
新聞中心
我是學院講師張敏，在學院 “4.20 IT充電節(jié)”(4月19~20日) 到來之際，和大家分享一下Spark Core之上擴建自己的模塊的經(jīng)驗。正文來啦~~~
創(chuàng)新互聯(lián)專注于企業(yè)全網(wǎng)整合營銷推廣、網(wǎng)站重做改版、阿瓦提網(wǎng)站定制設計、自適應品牌網(wǎng)站建設、H5建站、商城網(wǎng)站開發(fā)、集團公司官網(wǎng)建設、外貿(mào)網(wǎng)站制作、高端網(wǎng)站制作、響應式網(wǎng)頁設計等建站業(yè)務，價格優(yōu)惠性價比高，為阿瓦提等各大城市提供網(wǎng)站開發(fā)制作服務。
hbase-rdd是一個構(gòu)建在SparkContext基礎之上的用于對Hbase進行增刪改查的第三方開源模塊，目前***版本為0.7.1。目前該rdd在操作hbase時，默認調(diào)用隱式方法。
- implicitdef stringToBytes(s: String): Array[Byte] = {
- Bytes.toBytes(s)
- }
將RDD的key轉(zhuǎn)換成字節(jié)b，然后調(diào)用Hbase的put(b)方法保存rowkey，之后將RDD的每一行存入hbase。
在軌跡圖繪制項目數(shù)據(jù)計算中，我們考慮到hbase的rowkey的設計——盡量減少rowkey存儲的開銷。雖然hbase-rdd最終的rowkey默認都是采用字節(jié)數(shù)組，但這個地方我們希望按自己的方式組裝rowkey。使用MD5(imei)+dateTime組成的字節(jié)數(shù)組作為rowkey。因此默認的hbase-rdd提供的方法是不滿足我們存儲需求的，需要對源代碼進行修改。在toHbase方法中，有一個convert方法，該方法將對RDD中的每一行數(shù)據(jù)進行轉(zhuǎn)化，使用RDD中的key生成Put(Bytes.toBytes(key))對象，該對象為之后存儲Hbase提供rowkey。
在convert函數(shù)中，對其實現(xiàn)進行了改造，hbase-rdd默認使用stringToBytes隱式函數(shù)將RDD的String類型的key轉(zhuǎn)換成字節(jié)數(shù)組，這里我們需要改造，不使stringToBytes隱式方法，而是直接生成字節(jié)數(shù)據(jù)。
- 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)
- }
這樣就實現(xiàn)了使用自己的方式構(gòu)建rowkey，當然基于此思想我們可以使用任意的方式構(gòu)建rowkey。
在使用hbase-rdd插件的過程中，我在思考，默認的RDD上是沒有toHbase方法的，那為什么引入hbase-rdd包之后，RDD之上就有toHbase方法了?經(jīng)過查看源碼，發(fā)現(xiàn)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])
這兩個方法在發(fā)現(xiàn)RDD上沒有toHbase方法時會自動嘗試調(diào)用，從隱式定義中嘗試找到解決方案，嘗試之后發(fā)現(xiàn)有定義toHBaseRDDSimple隱式方法，于是調(diào)用該隱式方法新建HBaseWriteRDDSimple類，返回hBaseWriteRDDSimple，而在hBaseWriteRDDSimple對象中是有toHbase方法的，因此在引入hbase-rdd之后，可以發(fā)現(xiàn)原本沒有toHbase方法的RDD上有toHbase方法了。這一切都要歸功于Scala強大的隱式轉(zhuǎn)換功能。
那明白了原理，是否我們可以基于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")
- }
- }
項目結(jié)構(gòu)圖：
運行效果圖：
希望對大家以后的開發(fā)有幫助，同時借鑒本案例，在Spark Core之上構(gòu)建自己的小模塊。
學院 4.20 IT充電節(jié)
(19-20號兩天，100門視頻課程免單搶，更有視頻課程會員享6折，非會員享7折，套餐折上8折，微職位立減2000元鉅惠)
活動鏈接：http://edu./activity/lists/id-47.html?wenzhang
相關視頻教程：
【大數(shù)據(jù) Spark2.x 流數(shù)據(jù)處理】精通Spark流數(shù)據(jù)處理(持續(xù)完畢)
網(wǎng)站名稱：借hbase-rdd二次開發(fā)談如何在SparkCore之上擴建自己的模塊
網(wǎng)站路徑：http://www.5511xx.com/article/dphcisg.html
