无法在Spark上序列化任务

Question

我有一个这样的转变：

JavaRDD<Tuple2<String, Long>> mappedRdd = myRDD.values().map(
    new Function<Pageview, Tuple2<String, Long>>() {
      @Override
      public Tuple2<String, Long> call(Pageview pageview) throws Exception {
        String key = pageview.getUrl().toString();
        Long value = getDay(pageview.getTimestamp());
        return new Tuple2<>(key, value);
      }
    });

Pageview是一种：Pageview。JAVA

我在Spark上注册了这样的课程：

Class[] c = new Class[1];
c[0] = Pageview.class;
sparkConf.registerKryoClasses(c);

异常线程"main"org.apache.spark.SparkExctive：任务不能在org.apache.spark.util.ClosureCleaner$. ensureSerializable（ClosureCleaner.scala:166）在org.apache.spark.util.ClosureCleaner$.清洁（ClosureCleaner.scala:158）在org.apache.spark.SparkContext.clean（SparkContext.scala:1623）在org.apache.spark.rdd.RDD. map（RDD. scala： 286）在org.apache.spark.api.java.JavaRDD类似于$class.map（JavaRDDLike.scala:89）在org.apache.spark.api.java.AbstractJavaRDDLike.map（JavaRDDLike.scala:46）在org.apache.gora.tutorial.log.ExpleSpark.run（ExpleSpark.java:100）在org. apache. gora. tuorial. log。由java. io. NotSerializableExctive： org. apache. gora.教程. log.造成的。示例Spark序列化堆栈：-对象不可序列化（类： org. apache. gora.教程. log。示例Spark，值：org.apache.gora.tutorial.log.ExampleSpark@1a2b4497）-字段（类： org. apache. gora.教程. log.）-字段（类： org. apache. gora.教程. log。示例Spark1美元，名称： this0美元，类型：类org. apache. gora.教程. log。示例Spark）-对象（类org. apache. gora.教程. log。示例Spark$1@4ab2775d）-字段（类： org. apache. spark. api. java. JavaPairRDD$$anonfun$toScala功能1美元，名称： fun1美元，类型：接口org. apache. spark. appark

当我调试代码时，我看到JavaSerializer。scala被调用，即使有一个名为KryoSerializer的类。

PS 1：我不想使用Java序列化器，但是在Pageview上实现序列化器并不能解决问题。

PS 2：这并没有丢掉问题：

...
//String key = pageview.getUrl().toString();
//Long value = getDay(pageview.getTimestamp());
String key = "Dummy";
Long value = 1L;
return new Tuple2<>(key, value);
...

Answer 1

我在Java代码中多次遇到这个问题。虽然我使用的是Java序列化，但我会使包含该代码的类可序列化，或者如果您不想这样做，我会使该函数成为该类的静态成员。

下面是一个解决方案的代码片段。

public class Test {
   private static Function s = new Function<Pageview, Tuple2<String, Long>>() {

     @Override
     public Tuple2<String, Long> call(Pageview pageview) throws Exception {
       String key = pageview.getUrl().toString();
       Long value = getDay(pageview.getTimestamp());
       return new Tuple2<>(key, value);
      }
  };
}