aggregatebykey没有更新初始集的值

mitkmikd  于 2021-05-27  发布在  Hadoop
关注(0)|答案(1)|浏览(382)

中的值 hllTotal 更新了,但是 hllToday 每把钥匙都保持零。
有人能帮忙吗,为什么 hllToday 这里没有更新吗?

val hllToday: HllSerializable = new HllSerializable(new HLL(13, 5))
 val hllTotal: HllSerializable = new HllSerializable(new HLL(13, 5))

 val initialSet = (hllToday, hllTotal)
 val rdd7 = combinedRdd.map(tuple => {
      tuple.segmentId -> (tuple.entryTime, tuple.exitTime, tuple.hll)`
    }).aggregateByKey(initialSet)(
      (acc: (HllSerializable, HllSerializable), v: (Long, Long, HllSerializable)) => {
        if (v._1 == today_ts) {
          acc._1.getHll.union(v._3.getHll)
        }
        acc._2.getHll.union(v._3.getHll)
        acc
      }, (acc1: (HllSerializable, HllSerializable), acc2: (HllSerializable, HllSerializable)) =>    {
        acc1._1.getHll.union(acc2._1.getHll)
        acc1._2.getHll.union(acc2._2.getHll)
        acc1
      }
    )
jk9hmnmh

jk9hmnmh1#

显然,这个代码从未被调用:

if (v._1 == today_ts) {
    acc._1.getHll.union(v._3.getHll)
 }

尝试替换 if (v._1 == today_ts)if(true) 看看 hllToday 更新

相关问题