将不同的分组rdd值写入一个文件

Question

我有一个groupedrdd，其类型为key=string和value=iterable

值实际上以字符串格式保存JSON数据，而分组键的格式为 / /

示例：如果我的分组rdd中有以下键

tenant1/2016/12/output_data.json
tenant1/2017/01/output_data.json
tenant1/2017/02/output_data.json

那么在我的HDFS中我应该有三个文件

tenant1/2016/12/output_data.json
tenant1/2017/01/output_data.json
tenant1/2017/02/output_data.json

为此，我尝试了以下方法：

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]
}

groupedRDD.partitionBy(new HashPartitioner(1))
    .saveAsHadoopFile("/user/pkhode/output/", classOf[String], classOf[String], classOf[RDDMultipleTextOutputFormat])

/user/pkhode/output/tenant1/2016/12/output_data.json
/user/pkhode/output/tenant1/2017/01/output_data.json
/user/pkhode/output/tenant1/2017/02/output_data.json

List({json_object_in_string1}, {json_object_in_string2}, .....)

{json_object_in_string1}
{json_object_in_string2}
.....

groupedRDD.partitionBy(new HashPartitioner(1000)).mapValues(_.mkString("\n")).saveAsHadoopFile(outputPath, classOf[String], classOf[String], classOf[RDDMultipleTextOutputFormat])

java.lang.OutOfMemoryError: Requested array size exceeds VM limit
    at java.util.Arrays.copyOf(Arrays.java:2271)
    at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:113)
    at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
    at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:140)
    at com.esotericsoftware.kryo.io.Output.flush(Output.java:181)
    at com.esotericsoftware.kryo.io.Output.require(Output.java:160)
    at com.esotericsoftware.kryo.io.Output.writeString_slow(Output.java:462)
    at com.esotericsoftware.kryo.io.Output.writeString(Output.java:363)
    at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.write(DefaultSerializers.java:191)
    at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.write(DefaultSerializers.java:184)
    at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:628)
    at com.twitter.chill.TraversableSerializer$$anonfun$write$1.apply(Traversable.scala:29)
    at com.twitter.chill.TraversableSerializer$$anonfun$write$1.apply(Traversable.scala:27)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at com.twitter.chill.TraversableSerializer.write(Traversable.scala:27)
    at com.twitter.chill.TraversableSerializer.write(Traversable.scala:21)
    at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:628)
    at org.apache.spark.serializer.KryoSerializationStream.writeObject(KryoSerializer.scala:195)
    at org.apache.spark.serializer.SerializationStream.writeValue(Serializer.scala:135)
    at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
    at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)
    at org.apache.spark.scheduler.Task.run(Task.scala:86)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

当Spark将RDD保存为文本文件时，它只对RDD元素调用tostring。首先尝试将值映射到字符串:

rdd.mapValues(_.mkString("\n"))