借hbase-rdd二次开发谈如何在Spark Core之上扩建自己的模块
hbase-rdd是一个构建在SparkContext基础之上的用于对Hbase进行增删改查的第三方开源模块，目前最新版本为0.7.1。目前该rdd在操作hbase时，默认调用隐式方法。 implicitdefstringToBytes(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隐式方法，而是直接生成字节数据。 protecteddefconvert(id:String,values:Map[String,Map[String,A]],put:PutAdder[A])={ valstrs=id.split(",") valimei=strs{0} valdateTime=strs{1} valb1=MD5Utils.computeMD5Hash(imei.getBytes()) valb2=Bytes.toBytes(dateTime.toLong) valkey=b1.++(b2) valp=newPut(key)//改造 varempty=true for{ (family,content)<-values (key,value)<-content }{ empty=false if(StrUtils.isNotEmpty(family)&&StrUtils.isNotEmpty(key)){ put(p,family,key,value) } } if(empty)NoneelseSome(newImmutableBytesWritable,p) } 这样就实现了使用自己的方式构建rowkey，当然基于此思想我们可以使用任意的方式构建rowkey。 在使用hbase-rdd插件的过程中，我在思考，默认的RDD上是没有toHbase方法的，那为什么引入hbase-rdd包之后，RDD之上就有toHbase方法了?经过查看源码，发现hbase-rdd包中提供了两个隐式方法： implicitdeftoHBaseRDDSimple[A](rdd:RDD[(String,Map[String,A])])(implicitwriter:Writes[A]):HBaseWriteRDDSimple[A]=newHBaseWriteRDDSimple(rdd,pa[A]) implicitdeftoHBaseRDDSimpleTS[A](rdd:RDD[(String,Map[String,(A,Long)])])(implicitwriter:Writes[A]):HBaseWriteRDDSimple[(A,Long)]=newHBaseWriteRDDSimple(rdd,pa[A]) 这两个方法在发现RDD上没有toHbase方法时会自动尝试调用，从隐式定义中尝试找到解决方案，尝试之后发现有定义toHBaseRDDSimple隐式方法，于是调用该隐式方法新建HBaseWriteRDDSimple类，返回hBaseWriteRDDSimple，而在hBaseWriteRDDSimple对象中是有toHbase方法的，因此在引入hbase-rdd之后，可以发现原本没有toHbase方法的RDD上有toHbase方法了。这一切都要归功于Scala强大的隐式转换功能。 那明白了原理，是否我们可以基于RDD写自己的模块，说干就干! 第一步：新建Trait traitHaha{ implicitdefgaga[A](rdd:RDD[String]):Hehe= newHehe(rdd) } 第二步：新建Hehe类 finalclassHehe(rdd:RDD[String]){ defwow(tableName:String,family:String):Unit={ println("---------------------------------------------") println("tableName:"+tableName+"-family:"+family) println("size:"+rdd.count()) rdd.collect().foreach(data=>println(data)) println("---------------------------------------------") } } 第三步：新建包对象 packageobjecttestextendsHaha 第四步：新建test类 objectTest{ defmain(args:Array[String]){ valsparkConf=newSparkConf().setAppName("Test") valsc=newSparkContext(sparkConf) sc.makeRDD(Seq("one","two","three","four")).wow("taskDataPre","T") } } 项目结构图： 运行效果图： 希望对大家以后的开发有帮助，同时借鉴本案例，在Spark Core之上构建自己的小模块。 作者：张敏 来源：51CTO
