apache-spark - Apache Spark:按键将RDD对拆分为多个RDD以保存值

Question

我正在使用Spark 1.0.1处理大量数据。每行包含一个ID号，有些具有重复的ID。我想将所有具有相同ID号的行保存在同一位置，但是我很难高效地进行操作。我创建了一个(ID号，数据行)对的RDD [(String，String)]:

val mapRdd = rdd.map{ x=> (x.split("\\t+")(1), x)} 

一种有效但无效的方法是收集ID号，过滤每个ID的RDD，并将具有相同ID的值的RDD保存为文本文件。

val ids = rdd.keys.distinct.collect
ids.foreach({ id =>
    val dataRows = mapRdd.filter(_._1 == id).values
    dataRows.saveAsTextFile(id)
})

我还尝试了groupByKey或reduceByKey，以便RDD中的每个元组都包含一个唯一的ID号作为键，以及由该ID号的新行分隔的组合数据行的字符串。我只想通过一次foreach来遍历RDD一次以保存数据，但是它不能将值作为RDD给出

groupedRdd.foreach({ tup =>
  val data = sc.parallelize(List(tup._2)) //nested RDD does not work
  data.saveAsTextFile(tup._1)
})

本质上，我想通过一个ID号将一个RDD拆分为多个RDD，并将该ID号的值保存到自己的位置。

Answer 1

我认为这个问题类似于
Write to multiple outputs by key Spark - one Spark job

请引用那里的答案。

import org.apache.hadoop.io.NullWritable

import org.apache.spark._
import org.apache.spark.SparkContext._

import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any, Any] {
  override def generateActualKey(key: Any, value: Any): Any = 
    NullWritable.get()

  override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = 
    key.asInstanceOf[String]
}

object Split {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Split" + args(1))
    val sc = new SparkContext(conf)
    sc.textFile("input/path")
    .map(a => (k, v)) // Your own implementation
    .partitionBy(new HashPartitioner(num))
    .saveAsHadoopFile("output/path", classOf[String], classOf[String],
      classOf[RDDMultipleTextOutputFormat])
    spark.stop()
  }
}

刚刚在上面看到了类似的答案，但实际上我们不需要自定义分区。 MultipleTextOutputFormat将为每个键创建文件。可以将具有相同键的多个记录放入同一分区。

新的HashPartitioner(num)，其中num是所需的分区号。如果您有大量不同的键，可以将数字设置为big。在这种情况下，每个分区将不会打开太多的hdfs文件处理程序。