scala - Spark:RDD到列表

Question

我有一个RDD结构

RDD[(String, String)]

我想创建2个列表（rdd的每个维度一个）。

我尝试使用rdd.foreach（）并填充两个ListBuffers，然后将它们转换为Lists，但是我猜每个节点都会创建自己的ListBuffer，因为在迭代之后BufferLists为空。我该怎么做 ？

编辑：我的方法

val labeled = data_labeled.map { line =>
  val parts = line.split(',')
  (parts(5), parts(7))
}.cache()

var testList : ListBuffer[String] = new ListBuffer()

labeled.foreach(line =>
  testList += line._1
)
  val labeledList = testList.toList
  println("rdd: " + labeled.count)
  println("bufferList: " + testList.size)
  println("list: " + labeledList.size)

结果是：

rdd: 31990654
bufferList: 0
list: 0

Answer 1

如果您确实要创建两个列表（即您希望将所有分布式数据收集到驱动程序应用程序中（风险缓慢或OutOfMemoryError）），则可以使用collect，然后对结果使用简单的map操作：

val list: List[(String, String)] = rdd.collect().toList
val col1: List[String] = list.map(_._1)
val col2: List[String] = list.map(_._2)

或者，如果您想将RDD“拆分”为两个RDD，则无需收集数据，这非常相似：

rdd.cache() // to make sure calculation of rdd is not repeated twice
val rdd1: RDD[String] = rdd.map(_._1)
val rdd2: RDD[String] = rdd.map(_._2)

第三种选择是先映射到这两个RDD，然后收集它们中的每个，但与第一个选择并没有太大区别，并且存在相同的风险和局限性。