如何在 spark 中操作这个 Iterable？

Question

我有一个 rdd:

val input = RDD[(Int, Iterable[(info1, info2)])]

我需要的是：

RDD[(Int, Array[info2])]

现在我正在尝试

val output = input.mapValue(ele => ???.toArray)

???是我遇到的困难，我不知道如何访问 iterable 中的元组并且只将它们中的每一个都带到一个数组中。还是我选择了 mapValue 方法做错了什么？

Answer 1

如果结果不需要数组，并且可以使用任何类型的 Seq，也许可以使用以下内容：

给定这个数据集示例：

val inputrdd = sc.parallelize(Seq((1, Iterable((101,102))), (2, Iterable((201,202))), (3, Iterable((301,302)))))

您可以像这样使用 mapValue：

inputrdd.mapValues(iter => iter.map(_._2))

相当于：

inputrdd.map{case(k,iter)=>(k,iter.map(_._2))}

iter 是可迭代对象，它位于 map(_._2) 元组的第二部分。

至此，您现在了解如何访问和操作可迭代对象。如果你需要，强制一个数组，你可以这样做：

inputrdd.mapValues(iter => iter.map(_._2).toArray)

How to manipulate this Iterable in spark?