java - 使用 spark JAVA API takeOrdered 方法出错

Question

spark JavaRDDLike接口(interface)有api

 takeOrdered(int num, java.util.Comparator<T> comp)

或

 top(int num,java.util.Comparator<T> comp)

要返回前 k 个元素，但是当我使用带有 collect 方法的 wordcount 示例对其进行测试时，出现以下错误。

org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException

谁能帮帮我。非常感谢。

Answer 1

制作top()或 takeOrdered()工作，你必须确保你的自定义比较器类实现 java.io.Serializable .方法如下:这里我的 RDD 元素 (T) 是 Tuple2<String, Integer>我想要对第二个元组元素进行排序，它是一个整数:

import java.util.Comparator;
import java.io.Serializable;
....
static class MyTupleComparator implements 
   Comparator<Tuple2<String, Integer>>, Serializable {
   final static MyTupleComparator INSTANCE = new MyTupleComparator();
   // note that the comparison is performed on the key's frequency
   // assuming that the second field of Tuple2 is a count or frequency
   public int compare(Tuple2<String, Integer> t1, 
                      Tuple2<String, Integer> t2) {
       return -t1._2.compareTo(t2._2);    // sort descending
       // return t1._2.compareTo(t2._2);  // sort ascending
   }
}

然后通过MyTupleComparator.INSTANCE作为你的比较者。在 YARN 上运行 Spark 时我注意到的另一件事是你的 Spark 驱动程序应该实现 java.io.Serializable界面也是如此。

最好的问候，

马哈茂德·帕西安