您的位置：首页 > 其它

通过 “由对象Ｖ到对象Ｃ的转换” 来说明 Spark_Streaming api中 reduceByKey 与 combineByKey 注意事项

2016-08-11 08:51 701 查看

今年以来一直在学习通过Spark Streaming 来处理公司大数据相关的业务需求。去重＼汇总在大数据统计中是很常见的。而reduceByKey、combineByKey 在 Spark Streaming 中做合并操作时（由对象Ｖ到对象Ｃ的转换）很重要的两个api . [b]网上的事例大部分太过简单，或者讲解过于皮毛。[/b]

先对比下二者的函数签名：
class PairRDDFunctions[K, C](...) {
def reduceByKey(func: (C, C) => C): RDD[(K, C)]

def combineByKey[C](
createCombiner: V => C,
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C): RDD[(K, C)]
}

总体来看， reduceByKey 比 combineByKey 更简单，combineByKey 比 reduceByKey 更通用，另外，在实际转换（从对象Ｖ到对象Ｃ）过程当中，对开发员来讲是非常讲究的。而我将就实际的需求事例，来做讲解：

现有 V 与 C 两个对象：

classVobj(vId: String, vName: String, vType : String) extends Serializable{
var id:String = vId
val name:String = vName
val stype:String = vType
}

classCobj extends Serializable{

var lists = new ObjectArraySet[Vobj]()

}

现有数据内容如下：

val c01 = new Vobj("1","hu","1")
val c02 = new Vobj("2","hu","2")
val c03 = new Vobj("3","hu","3")
val c04 = new Vobj("4","hu","1")
val c05 = new Vobj("5","hu","1")

现在需要按Ｖ对象的值进行分组合并，将转化成Ｃ对象，现有两种实现方式：

第一种：

val testCobjRdd = sc.parallelize(List(c01,c02,c03,c04,c05))
testCobjRdd.map(x => {
//在map阶段就进行对象的转换
val cj = new Cobj()
cj.lists.add(x)
(x.stype,cj)
}).reduceByKey((pre:Cobj,aft:Cobj) => {
//构造一个新的中间对象并间数据进行汇集
val mid = new Cobj()
mid.lists.addAll(pre.lists)
mid.lists.addAll(aft.lists)
mid
}).foreach(x => {
println(x._1+"-"+x._2.lists.size())
})

第二种：

testCobjRdd.map(x => {
(x.stype,x)
}).combineByKey[Cobj]((v:Vobj)=>{
//创建Ｖ对象的初始化[在第一个RDD中的第一条Ｃ记录中逻辑]
val midC = new Cobj()
midC.lists.add(v)
midC
},(c:Cobj,v:Vobj) => {
//由Ｃ到Ｖ汇集业务逻辑[在第一个RDD中的第一条Ｖ记录与第二条Ｃ记录的逻辑]
c.lists.add(v)
c
},(cPre:Cobj,cAft:Cobj) => {
//由Ｖ到Ｖ汇集业务逻辑[在第一个与第二个RDD中汇集时的逻辑]
val midC = new Cobj()
midC.lists.addAll(cPre.lists)
midC.lists.addAll(cAft.lists)
midC
}).foreach(x => {
println(x._1+"-"+x._2.lists.size())
})

结果都会如下：

2-1

3-1

1-3

其中，这两种实现逻辑是等价的。但是根据实际消耗的时间来看，第二种要好于第一种。

另外，重点讲一个严重错误事例，有的童鞋为了省事，只用Ｖ一个对象做合并，然后Ｖ对象内新建一个集合属性，如下：

classVobj(vId: String, vName: String, vType : String) extends Serializable{
var id:String = vId
val name:String = vName
val stype:String = vType

var lists = new ObjectArraySet[Vobj]()
}

然后，在代码中如下：

val testCobjRdd = sc.parallelize(List(c01,c02,c03,c04,c05))
testCobjRdd.map(x => {
(x.stype,x)
}).reduceByKey((pre,aft) => {
val cMid = new Vobj(null,null,null)
cMid.lists.add(pre)
cMid.lists.add(aft)
cMid
}).foreach(x => {
println(x._1+"-"+x._2.lists.size())
})

却发现结果为：

2-0

3-0

1-2

所以，上面的这种事例是严重错误的。因为，如果结果集中如果只有一个对象，那reduce的时候就不会参于里面的函数运算。

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签： reduceByKey combineByKey spark

相关文章推荐

新的分享

章节导航