1

我想导入一个 PMML 模型，以使用 Spark 计算分数。当我不使用火花时一切正常，但我不能在映射器中使用我的方法。

问题是我需要一个来自 org.jpmml.evaluator.Evaluator 的评估对象，它似乎不是可序列化的。因此，我尝试使用以下课程使其成为可串行化：

package util;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;

import org.jpmml.evaluator.Evaluator;

public class SerializableEvaluator implements Serializable {

    private static final long serialVersionUID = 6631604036553063657L;
    private Evaluator evaluator;

    public SerializableEvaluator(Evaluator evaluator) {
        this.evaluator = evaluator;
    }

    public Evaluator getEvaluator() {
        return evaluator;
    }

    private void writeObject(ObjectOutputStream out) throws IOException {
        out.writeObject(evaluator);
    }

    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        Evaluator eval = (Evaluator) in.readObject();
    }
}

我还使我所有的类都可序列化。

这是我的代码示例：

        logger.info("Print 5 first rows----------------------------");
        strTitanicRDD
                .take(5)
                .forEach(row -> logger.info(row));
        logger.info("Print 5 first Titatnic Obs---------------------");
        strTitanicRDD
                .map(row -> new TitanicObservation(row))
                .take(5)
                .forEach(titanic -> logger.info(titanic.toString()));
        logger.info("Print 5 first Scored Titatnic Obs---------------");

        try{strTitanicRDD
            .map(row -> new TitanicObservation(row))
            .map(
                new Function<TitanicObservation,String>(){

                    private static final long serialVersionUID = -2968122030659306400L;

                    @Override
                    public String call(TitanicObservation titanic) throws Exception {
                        String res = PmmlUtil.computeScoreTitanic(evaluator, titanic);
                        return res;
                    }

                })
        .take(5)
        .forEach(row -> logger.info(row));

但我不认为我的代码会帮助你解决我的问题，这很清楚（见日志：）

org.apache.spark.SparkException：在 org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) 的 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166) 任务不可序列化) 在 org.apache.spark.SparkContext.clean(SparkContext.scala:1623) 在 org.apache.spark.rdd.RDD.map(RDD.scala:286) 在 org.apache.spark.api.java.JavaRDDLike$ class.map(JavaRDDLike.scala:89) 在 org.apache.spark.api.java.AbstractJavaRDDLike.map(JavaRDDLike.scala:46) 在 score.acv.AppWithSpark.main(AppWithSpark.java:117) 在 sun.reflect .NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:497) 在 org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:577)在 org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:174) 在 org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:197) 在 org.apache.spark.deploy。 SparkSubmit$.main(SparkSubmit.scala:112) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)197) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:112) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)197) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:112) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

引起：java.io.NotSerializableException：org.xml.sax.helpers.LocatorImpl 序列化堆栈：

    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:38)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:80)
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
    ... 15 more
4

1 回答 1

2

在接口后面org.jpmml.evaluator.Evaluator有一个org.jpmml.evaluator.ModelEvaluator子类的实例。该类ModelEvaluator及其所有子类在设计上都是可序列化的。问题与您在开始org.dmg.pmml.PMML时提供给方法的对象实例有关ModelEvaluatorFactory#newModelManager(PMML)。

简而言之，每个 PMML 类模型对象都可以附加 SAX 定位器信息。这在开发和测试阶段用于定位有问题的 XML 内容非常有用。但是，在生产阶段，不应再保留此信息。您可以通过正确配置 JAXB 运行时禁用 SAX Locator 信息，或者通过PMMLObject#setLocator(Locatable)使用null参数调用简单地清除现有 SAX Locator 实例。后一个功能由org.jpmml.model.visitors.LocatorNullifierVisitor 类正式化。

有关完整示例，请参阅官方JPMML-Spark 项目org.jpmml.spark.EvaluatorUtil的实用程序类（尤其是第 73 到 75 行）。为什么不首先使用 JPMML-Spark？

于 2016-02-17T20:14:12.787 回答