代码之家 › 专栏 › 技术社区 › pirox22

元组可数值的约化基

rdd apache-spark scala

2

pirox22 · 技术社区 · 7 年前

我试着计算特定日期特定物品的出现次数。

Date\tItem1:AppearencesOfItem1,...,ItemN:AppearencesOfItemN

20/10/2000\tItem1:1,Item2:5
20/10/2000\tItem1:2
21/10/2000\tItem1:5

为此,我创建了以下pairdd结构:

[(20/10/2000, (Item1, 1))
(20/10/2000, (Item2, 5))
(20/10/2000, (Item1, 5))
(21/10/2000, (Item1, 5))]

groupByKey 在导致:

[(20/10/2000, Iterable[(Item1, 1), (Item2, 5), (Item1, 5))
 (21/10/2000, Iterable[(Item1, 5)]

在这一步之后,我要做的是减少这些对的值,并对共享同一个键的项的外观求和,这样结果将变成:

[(20/10/2000, Iterable[(Item1, 6), (Item2, 5))
 (21/10/2000, Iterable[(Item1, 5)]

然而,我还没有找到一种方法来降低这些pairdds的价值。我的方法一开始就错了吗?

2 回复 | 直到 7 年前

1

2

pasha701 7 年前

按前两列求和

按第一列分组

val data = List( 
  ("20/10/2000", "Item1", 1),
  ("20/10/2000", "Item2", 5),
  ("20/10/2000", "Item1", 5),
  ("21/10/2000", "Item1", 5)
)
val originalRDD = sparkContext.parallelize(data)

val sumRDD = originalRDD.map(v => ((v._1, v._2), v._3)).reduceByKey(_ + _)
sumRDD.map(v => ((v._1._1), (v._1._2, v._2))).groupByKey().foreach(println)

输出:

(21/10/2000,CompactBuffer((Item1,5)))
(20/10/2000,CompactBuffer((Item1,6), (Item2,5)))

2

1

SCouto 7 年前

希望这能有所帮助,这可能不是他最优雅的方式,但似乎符合您的要求:

rdd.groupByKey.mapValues(x => x.groupBy(_._1).mapValues(x => x.map(_._2).sum))

首先将值映射到groupbyitemid,然后在分组列表上,再次映射值以仅保留第二个元素(整数),这样就可以直接求和

scala> rdd.groupByKey.mapValues(x => x.groupBy(_._1).mapValues(x => x.map(_._2).sum)).foreach(println)
(21/10/2000,Map(Item1 -> 5))
(20/10/2000,Map(Item2 -> 5, Item1 -> 6))

我在你的RDD里创建了一个地图,如果你想让它成为列表或者其他什么的话,只需要执行一个 toList

rdd.groupByKey.mapValues(x => x.groupBy(_._1).mapValues(x => x.map(_._2).sum).toList)