5

我们需要有效地转换大量键/值对列表，如下所示：

val providedData = List(
        (new Key("1"), new Val("one")),
        (new Key("1"), new Val("un")),
        (new Key("1"), new Val("ein")),
        (new Key("2"), new Val("two")),
        (new Key("2"), new Val("deux")),
        (new Key("2"), new Val("zwei"))
)

进入每个键的值列表，如下所示：

val expectedData = List(
  (new Key("1"), List(
    new Val("one"), 
    new Val("un"), 
    new Val("ein"))),
  (new Key("2"), List(
    new Val("two"), 
    new Val("deux"), 
    new Val("zwei")))
)

键值对来自大型键/值存储（Accumulo），因此键将被排序，但通常会跨越 Spark 分区边界。每个键可以有数百万个键和数百个值。

我认为这项工作的正确工具是 spark 的 combineByKey 操作，但只能找到具有泛型类型（如 Int）的简洁示例，我无法将其推广到上述用户定义的类型。

由于我怀疑许多其他人会有同样的问题，我希望有人可以提供完全指定（详细）和简洁的 scala 语法示例，以便将 combineByKey 与上述用户定义类型一起使用，或者可能指出一个更好的工具我错过了。

4

1 回答 1

4

我不是真正的 Spark 专家，但基于这个问题，我认为您可以执行以下操作：

val rdd = sc.parallelize(providedData)

rdd.combineByKey(
    // createCombiner: add first value to a list
    (x: Val) => List(x),
    // mergeValue: add new value to existing list
    (acc: List[Val], x) => x :: acc,
    // mergeCominber: combine the 2 lists
    (acc1: List[Val], acc2: List[Val]) => acc1 ::: acc2
)

使用aggregateByKey：

rdd.aggregateByKey(List[Val]())(
    (acc, x) => x :: acc,
    (acc1, acc2) => acc1 ::: acc2
)
于 2015-06-20T12:17:59.083 回答