合并不会减少输出文件的数量

Question

我有一个火花作业，它管理HDFS上的RDD[专业记录库]。

我的问题是它会生成很多文件，包括 95% 的空 avro 文件。我尝试使用合并来减少RDD上的分区数量，从而减少输出文件的数量，但它没有效果。

 def write(data: RDD[SpecificRecordBase]) = {
   data.coalesce(1, false)    //has no effect
   val conf = new Configuration()
   val job = new org.apache.hadoop.mapreduce.Job(conf)

   AvroJob.setOutputKeySchema(job, schema)
   val pair = new PairRDDFunctions(rdd)
   pair.saveAsNewAPIHadoopFile(
     outputAvroDataPath,
     classOf[AvroKey[SpecificRecordBase]],
     classOf[org.apache.hadoop.io.NullWritable],
     classOf[AvroKeyOutputFormat[SpecificRecordBase]],
     job.getConfiguration)
}

我想rdd分区配置和HDFS分区之间丢失了一些东西，也许SaveAsNewAPIHadoopFile没有考虑到它，但我不确定。

我错过了什么吗？

有人能解释一下根据rdd分区调用SaveAsNewAPIHadoopFile时真正附加的内容吗？

Answer 1

感谢@0x0FFF回答了我自己的问题，正确的代码应该是：

    def write(data: RDD[SpecificRecordBase]) = {
           val rdd = data.map(t => (new AvroKey(t), org.apache.hadoop.io.NullWritable.get))
           val rdd1Partition = rdd.coalesce(1, false)  //change nb of partitions to 1

           val conf = new Configuration()
           val job = new org.apache.hadoop.mapreduce.Job(conf)

           AvroJob.setOutputKeySchema(job, schema)
           val pair = new PairRDDFunctions(rdd1Partition) //so only one file will be in output
           pair.saveAsNewAPIHadoopFile(
             outputAvroDataPath,
             classOf[AvroKey[SpecificRecordBase]],
             classOf[org.apache.hadoop.io.NullWritable],
             classOf[AvroKeyOutputFormat[SpecificRecordBase]],
             job.getConfiguration)
        }

再次感谢！