将 RDD 的前 k 个值写入 Spark 中的文件答案

【问题标题】：Write top k values of RDD to a file in Spark将 RDD 的前 k 个值写入 Spark 中的文件
【发布时间】：2015-12-07 21:33:05
【问题描述】：

我想从 Tuple2 对象的 RDD 中提取前 k 个值并将它们写入文件。我一直在尝试的方法是使用 top 函数并将其传递给 Comparator 对象。这将返回一个 Tuple2 对象列表。

现在我可以在单个节点上合并此列表并遍历此列表并使用 FileWriter 将值打印到文件中。但这将导致一个文件。而且我希望每个 reducer 有几个文件。

有什么方法可以在 List 而不是 RDD 上使用 saveAsTextFile？我正在使用带有 Java 的 Spark 1.5.1。

Comparator<Tuple2<String, Double>> tc = new TupleComparator();

    List<Tuple2<String,Double>> output = ranks.coalesce(1,true).top(topK,tc);

  /*  for (Tuple2<String,Double> tuple : output){
        System.out.println(tuple._1() + " has rank: " + tuple._2() + ".");
    }*/
    BufferedWriter writer = new BufferedWriter(new FileWriter(args[2],false));
    int i=1;
    for(Tuple2<String,Double> tuple: output) {
      writer.write(i + " " + tuple._1 + " " + tuple._2);
      writer.newLine();
      i++;
    }
    writer.close();

【问题讨论】：

你应该以a Minimal, Complete, and Verifiable example开头。

标签： java apache-spark rdd

【解决方案1】：

您可以使用foreachPartition，并在每个分区上使用 FileWriter（或我的代码示例中的 PrintWriter），如下所示：

import scala.util.Random

sc.parallelize( {1 to 12}.zipWithIndex, 4)
      .foreachPartition( part => 
            printToFile(new File(s"partition_${Random.nextInt}"))  
            {p=> part.foreach(p.println)}
       )

我使用以下内容输出列表：

import java.io._

  //  usage:  printToFile(new File("filename")) { p => myCollection.foreach(p.println) }

def printToFile(f: java.io.File)(op: java.io.PrintWriter => Unit) { 
        val p = new java.io.PrintWriter(f); 
        try { op(p) } finally { p.close() } 
}

【讨论】：