按键Spark写入多个输出-一个Spark作业

Question

如何在单个作业中使用Spark根据密钥写入多个输出。

相关：按键写入多个输出扩展Hadoop，一个MapRe员作业

例如。

sc.makeRDD(Seq((1, "a"), (1, "b"), (2, "c")))
.writeAsMultiple(prefix, compressionCodecOption)

将确保cat前缀/1为

a
b

而cat prefix/2将是

c

编辑：我最近添加了一个新的答案，包括完整的导入，皮条客和压缩编解码器，请参阅https://stackoverflow.com/a/46118044/1586965，这可能是有帮助的，除了早期的答案。

Answer 1

如果一个给定的密钥可能有很多值，我认为可伸缩的解决方案是每个分区为每个密钥写一个文件。不幸的是，Spark中没有内置的支持，但我们可以激发一些东西。

sc.makeRDD(Seq((1, "a"), (1, "b"), (2, "c")))
  .mapPartitionsWithIndex { (p, it) =>
    val outputs = new MultiWriter(p.toString)
    for ((k, v) <- it) {
      outputs.write(k.toString, v)
    }
    outputs.close
    Nil.iterator
  }
  .foreach((x: Nothing) => ()) // To trigger the job.

// This one is Local, but you could write one for HDFS
class MultiWriter(suffix: String) {
  private val writers = collection.mutable.Map[String, java.io.PrintWriter]()
  def write(key: String, value: Any) = {
    if (!writers.contains(key)) {
      val f = new java.io.File("output/" + key + "/" + suffix)
      f.getParentFile.mkdirs
      writers(key) = new java.io.PrintWriter(f)
    }
    writers(key).println(value)
  }
  def close = writers.values.foreach(_.close)
}

（用您选择的分布式文件系统操作替换PrintWriter。）

这使得RDD单次通过，并且不执行洗牌。它为每个键提供一个目录，每个目录中包含许多文件。

Answer 2

我会这样做，这是可扩展的

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将为每个键创建文件。相同键的多个记录落入同一分区是可以的。

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

Answer 3

如果使用Spark 1.4，由于DataFrame API，这将变得非常容易。（数据帧是在Spark 1.3中引入的，但我们需要的partitionBy（）是在1.4中引入的。）

如果您开始使用RDD，首先需要将其转换为DataFrame：

val people_rdd = sc.parallelize(Seq((1, "alice"), (1, "bob"), (2, "charlie")))
val people_df = people_rdd.toDF("number", "name")

在Python中，同样的代码是：

people_rdd = sc.parallelize([(1, "alice"), (1, "bob"), (2, "charlie")])
people_df = people_rdd.toDF(["number", "name"])

一旦有了一个数据帧，就可以根据一个特定的键写入多个输出。更重要的是——这就是DataFrame API的美妙之处——Python、Scala、Java和R的代码几乎相同：

people_df.write.partitionBy("number").text("people")

如果需要，您可以轻松使用其他输出格式：

people_df.write.partitionBy("number").json("people-json")
people_df.write.partitionBy("number").parquet("people-parquet")

在这些示例中，Spark将为我们已在其中分区DataFrame的每个键创建一个子目录：

people/
  _SUCCESS
  number=1/
    part-abcd
    part-efgh
  number=2/
    part-abcd
    part-efgh