I am trying to load a PMML file in a spark job. Instantiate it only once and pass it to the executors. But I get a NotSerializableException for org.xml.sax.helpers.LocatorImpl which is used inside jpmml.
I have this class Prediction.java: public class Prediction implements Serializable { private RegressionModelEvaluator rme; public Prediction() throws Exception { InputStream is = .....getResourceAsStream("model.pmml"); Source source = ImportFilter.apply(new InputSource(is)); PMML model = JAXBUtil.unmarshalPMML(source); rme = new RegressionModelEvaluator(model); is.close(); } public Map predict(params) { ...... return rme.evaluate(params); } } Now I want to instantiate it only once since the "JAXBUtil.unmarshalPMML(source)" step takes about 2-3seconds. It works fine I instantiate inside the map{} So I do this in my driver: Prediction prediction = new Prediction(); JavaRDD<String> result = rdd1 .cartesian(rdd2) .map(t -> {...<need to use "prediction" here ...}); But when I do that, I get this error: Caused by: java.io.NotSerializableException: org.xml.sax.helpers.LocatorImpl Serialization stack: - object not serializable (class: org.xml.sax.helpers.LocatorImpl, value: org.xml.sax.helpers.LocatorImpl@6bce4140) - field (class: org.dmg.pmml2_7.PMMLObject, name: locator, type: interface org.xml.sax.Locator) - object (class org.dmg.pmml2_7.PMML, org.dmg.pmml2_7.PMML@722531ab) - field (class: org.jpmml.evaluator.PMMLManager, name: pmml, type: class org.dmg.pmml2_7.PMML) - object (class org.jpmml.evaluator.RegressionModelEvaluator, org.jpmml.evaluator.RegressionModelEvaluator@4c24f3a2) - field (class: com.x.ds.util.Predict, name: rme, type: class org.jpmml.evaluator.RegressionModelEvaluator) - object (class com.x.ds.util.Predict, com.x.ds.util.Predict@28f154cc) - element of array (index: 1) - array (class [Ljava.lang.Object;, size 2) - field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;) - object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class runner.SparkSimRunner, functionalInterfaceMethod=org/apache/spark/api/java/function/Function.call:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeSpecial runner/SparkSimRunner.lambda$runSimulations$ad93f7a8$1:(Lcom/x/ds/util/Predict;Lscala/Tuple2;)Ljava/util/List;, instantiatedMethodType=(Lscala/Tuple2;)Ljava/util/List;, numCaptured=2]) - writeReplace data (class: java.lang.invoke.SerializedLambda) - object (class runner.SparkSimRunner$$Lambda$31/1961138094, runner.SparkSimRunner$$Lambda$31/1961138094@782fd504) - field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function) - object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, <function1>) at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47) at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81) at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:312) I am doing this right? -- Thanks, -Utkarsh