scala - 如何使用组合键？

Question

我正在尝试使用 combineByKey 获得相同的 countByKey 结果。

scala> ordersMap.take(5).foreach(println)
(CLOSED,1)
(PENDING_PAYMENT,2)
(COMPLETE,3)
(CLOSED,4)
(COMPLETE,5)

这是我的输入，我想使用 combineByKey 来获取 countByKey 的输出。

countByKey 的输出(正确)

PAYMENT_REVIEW 729
CLOSED 7556
SUSPECTED_FRAUD 1558
PROCESSING 8275
COMPLETE 22899
PENDING 7610
PENDING_PAYMENT 15030
ON_HOLD 3798
CANCELED 1428

我用过 combineByKey

val combine = ordersMap.combineByKey(  x => 1 , (a:Int ,v) => a +1 , (a : Int,v : Int) => a +1  )

但我得到了意想不到的结果，我不确定为什么。

combineByKey 的结果

(PENDING_PAYMENT,7600)
(CLOSED,3878)
(CANCELED,699)
(PAYMENT_REVIEW,368)
(PENDING,3764)
(ON_HOLD,1896)
(PROCESSING,4100)
(SUSPECTED_FRAUD,773)
(COMPLETE,11372)

谢谢

Answer 1

那是因为你错误地应用了函数。您传递给 combineByKey 的最后一个函数需要合并可能在不同执行器上计算的累加器类型 (C) 的两个值。这就是函数名称被称为 mergeCombiners 的原因。

文档:

combineByKey[C](createCombiner: (V) ⇒ C, 
                mergeValue: (C, V) ⇒ C, 
                mergeCombiners: (C, C) ⇒ C): RDD[(K, C)]

你应该如何应用 mergeCombiners:

val combine = 
  ordersMap.combineByKey(_ => 1 , (a: Int, _) => a + 1, (a: Int, v: Int) => a + v)