spark combineByKey 代替 groupByKey

combineByKey 和 groupByKey 的区别

两者都可以对数据进行分组统计，但是前者在一般情况下，进行数据混洗的数据会远远小于后者。前者部分数据会在excutor端进行本地合并，然后将合并之后的数据进行网络混洗，后者是全部的数据进行网络混洗。

项目中的例子需求

hbase 中存储了车辆的 行驶数据，包括 车辆id，日期，当天行驶的里程数据，每天急加速，急减速的次数等
现有一个需求，计算每一个车辆的平均速度，和每一个月的平均行驶天数，里程的中位数等。
数据量在上亿的级别。数据hbase进行预分区， rowkey为 车辆id的反转加上日期 ："10924_20191010"

解决方案

第一版的解决方案为直接 读取hbase 数据进行foreachPartition 操作。因为hbase进行的预分区。相同的车辆id必然在同一个partition里面，但是，发现线上hbase region进行自动划分。但是有极少数的车辆id不在同一个region里面。此方案不行

第二版，直接读取数据进行全局的groupByKey操作。测试环境5千万条数据，跑的时间很慢。数据混洗时间较长。网络混洗的数据量较大。

第三版，为了优化网络混洗。combineByKey  代替groupByKey。 测试环境跑5千万条数据，不到3分钟跑完。

关于 combineByKey 的比较好的例子

https://backtobazics.com/big-data/apache-spark-combinebykey-example/](https://backtobazics.com/big-data/apache-spark-combinebykey-example/)

部分代码

case class VehicleMileage(vehicleId: Int, mileage: Int, reportTime: String)

 def createCombiner(v: VehicleMileage): List[VehicleMileage] = {
    List(v)
  }

 def mergeValue(vList: List[VehicleMileage], v: VehicleMileage): List[VehicleMileage] = {
    v :: vList
 }

 def mergeCombiners(vList1: List[VehicleMileage], vList2: List[VehicleMileage]): List[VehicleMileage] = {
    vList1 ::: vList2
 }

val hbaseConfig = HBaseConfiguration.create();
val scan = new Scan()
scan.addFamily("F".getBytes())
scan.setCaching(1000)
scan.setCacheBlocks(false)
hbaseConfig.set("hbase.zookeeper.quorum", zkUrl);
hbaseConfig.set("zookeeper.znode.parent", "/hbase-unsecure");
hbaseConfig.set(TableInputFormat.INPUT_TABLE, "xxxxxxxxx")
hbaseConfig.set(TableInputFormat.SCAN, TableMapReduceUtil.convertScanToString(scan))
val hbaseRdd = sc.newAPIHadoopRDD(hbaseConfig,
classOf[TableInputFormat],
classOf[ImmutableBytesWritable],
     classOf[Result])

hbaseRdd.combineByKey(createCombiner, mergeValue, mergeCombiners).foreachPartition(parition => {
    //分组之后计算 统计指标， 平均速度等
})

最后编辑于：2020.01.13 16:47:24