hadoop - 如何使用具有 ((key1,value),(key2,value)) 模式的 Apache Spark 查找键的所有值的总和

Question

我有一个数据集如下-

A  B  C
(a,c,30)
(a,b,20)
(b,c,10)
(c,d,1)

现在我需要处理上述数据以获得如下输出 -A 列中的任何键都将乘以 C 的 2 倍B 列中的任何键都将乘以 C 的 3 倍

所以这里的预期输出将是 -

a   100      =30*2+20*2
b   80       =20*3+10*2
c   122      =30*3+10*3+1*2
d   3        =1*3 

我可以像下面这样写-

val x = sc.parallelize(List(
  ("a","b",20),
  ("b","c",10),
  ("a","c",30),
  ("c","d",1)
))

val myVal = x.map({
  case (a,b,c) => ((a-> 2 * c), (b -> 3 * c))
})

myVal.foreach(println)

输出-

((a,60),(c,90))
((c,2),(d,3))
((a,40),(b,60))
((b,20),(c,30))

在那之后我无法进一步打破它如何使用 spark scala 获得预期的结果？

Answer 1

重点是先让它变平——将一个值与一个键相关联。然后就可以使用reduceByKey操作来总结了。

我不是 scala 开发人员，但像这样的东西可能会起作用。

myVal
  .flatMap({ case (a, b, c) => List(a -> 2 * c, b -> 3 * c) })
  .reduceByKey((a, b) => a + b)
  .foreach(println(_))

此处列出的是每次都必须创建的附加对象，最好避免使用它。所以，类似这样的事情可能会奏效 - 查看数据两次，但之前缓存它。

myVal.cache()
  .map({ case (a, b, c) => a -> 2 * c })
  .union(rdd.map({ case (a, b, c) => b -> 3 * c }))
  .reduceByKey((a, b) => a + b)
  .foreach(println(_))
myVal.unpersist()