Spark过滤器并多次计算大RDD

Question

假设我有一个RDD[(String，Int)]，如下例所示:

(A, 0)
(B, 0)
(C, 1)
(D, 0)
(E, 2)
(F, 1)
(G, 1)
(H, 3)
(I, 2)
(J, 0)
(K, 3)

我想有效地打印包含 0、1、2 等的记录总数。由于RDD包含数百万个条目，我希望尽可能高效地做到这一点。

此示例的输出将返回如下内容：

Number of records containing 0 = 4
Number of records containing 1 = 3
Number of records containing 2 = 2
Number of records containing 3 = 2

目前，我尝试通过对大RDD执行过滤器，然后分别对0、1、2、..执行count（）来实现这一点。我正在使用Scala。

有没有更高效的方法来做到这一点？我已经缓存了RDD，但是我的程序仍然耗尽了内存(我已经将驱动内存设置为5G)。

编辑：正如Tzach所建议的，我现在使用CountByKey：

rdd.map(_.swap).countByKey()

我是否可以通过将字符串值更改为一个元组(其中第二个元素是“m”或“f”)，然后获得该元组的第二个元素的每个唯一值的每个键的计数来改进这一点？

例如：

(A,m), 0)
(B,f), 0)
(C,m), 1)
(D,m), 0)
(E,f), 2)
(F,f), 1)
(G,m), 1)
(H,m), 3)
(I,f), 2)
(J,f), 0)
(K,m), 3)

会导致

((0,m), 2)
((0,f), 2)
((1,m), 2)
((1,f), 1)
((2,m), 0)
((2,f), 2)
((3,m), 2)
((3,f), 0)

提前感谢！

Answer 1

您可以使用方便的 countByKey 来做到这一点 - 只需事先交换输入中的位置即可使数值成为键：

val rdd = sc.parallelize(Seq(
  ("A", 0), ("B", 0), ("C", 1), ("D", 0), ("E", 2),
  ("F", 1), ("G", 1), ("H", 3), ("I", 2), ("J", 0), ("K", 3)
))

rdd.map(_.swap).countByKey().foreach(println)
// (0,4)
// (1,3)
// (3,2)
// (2,2)

编辑：＜code＞countByKey＜/code＞完全按照它听起来的样子执行-所以无论你想使用什么键，只要将你的RDD转换为元组的左侧部分，例如：

rdd.map { case ((a, b), i) => ((i, b), a) }.countByKey()

或：

rdd.keyBy { case ((_, b), i) => (i, b) }.countByKey()