R で生成された PMML モデル ファイルを Spark コンテキストにインポートし、それを使用してスコアを予測しようとしています。これは、Spark で使用されるコードです。
JavaRDD<String> scoreData = data.map(new Function<String, String>() {
@Override
public String call(String line) throws Exception {
String[] row = line.split(",");
PMML pmml;
Evaluator evaluator;
FileSystem fs = FileSystem.get(new Configuration());
FSDataInputStream inStr = fs.open(new Path("PATH_TO_PMML_FILE"));
Source transformedSource = ImportFilter.apply(new InputSource(inStr));
pmml = JAXBUtil.unmarshalPMML(transformedSource);
System.out.println(pmml.getModels().get(0).getModelName());
ModelEvaluatorFactory modelEvaluatorFactory = ModelEvaluatorFactory.newInstance();
ModelEvaluator<?> modelEvaluator = modelEvaluatorFactory.newModelManager(pmml);
System.out.println(modelEvaluator.getSummary());
evaluator = (Evaluator) modelEvaluator;
List<FieldName> activeFields = evaluator.getActiveFields();
double[] features = new double[row.length - 2]; // row - {contact_id,label}
StringBuilder strBld = new StringBuilder();
Map<FieldName, FieldValue> arguments = new LinkedHashMap<FieldName, FieldValue>();
strBld.append(row[0]);
for (int i = 3; i <= row.length - 1; i++) {
//from f1 - f16
FieldValue activeValue = evaluator.prepare(activeFields.get(i - 3), Double.parseDouble(row[i]));
arguments.put(activeFields.get(i - 3), activeValue);
}
}
コア Java 環境 (Spark コンテキストなし) で実行すると、コードは正常に機能しましたが、上記のコードを実行すると、次の例外が発生します。
java.lang.NoSuchMethodError: com.google.common.collect.Range.closed(Ljava/lang/Comparable;Ljava/lang/Comparable;)Lcom/google/common/collect/Range;
at org.jpmml.evaluator.Classification$Type.<clinit>(Classification.java:278)
at org.jpmml.evaluator.ProbabilityDistribution.<init>(ProbabilityDistribution.java:26)
at org.jpmml.evaluator.GeneralRegressionModelEvaluator.evaluateClassification(GeneralRegressionModelEvaluator.java:333)
at org.jpmml.evaluator.GeneralRegressionModelEvaluator.evaluate(GeneralRegressionModelEvaluator.java:107)
at org.jpmml.evaluator.ModelEvaluator.evaluate(ModelEvaluator.java:266)
at org.zcoe.spark.pmml.PMMLSpark_2$1.call(PMMLSpark_2.java:146)
at org.zcoe.spark.pmml.PMMLSpark_2$1.call(PMMLSpark_2.java:1)
at org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1.apply(JavaPairRDD.scala:999)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$17.apply(RDD.scala:813)
at org.apache.spark.rdd.RDD$$anonfun$17.apply(RDD.scala:813)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1503)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1503)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
at org.apache.spark.scheduler.Task.run(Task.scala:64)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
問題は、コードの実行に必要な Guvava Jar ファイルの互換性にあるようです。com.google.common.collect.Range クラスを含むすべての Jar を Spark のクラス パスから削除しましたが、同じ問題が引き続き発生します。
Spark ジョブの詳細は以下のとおりです。
spark-submit --jars ./lib/pmml-evaluator-1.2.0.jar,./lib/pmml-model-1.2.2.jar,./lib/pmml-manager-1.1.20.jar,./ lib/pmml-schema-1.2.2.jar,./lib/guava-15.0.jar --class
[ステージ 0:> (0 + 2) / 2]15/06/26 14:39:15 エラー YarnScheduler: hslave2 のエグゼキューター 1 が失われました: リモート Akka クライアントの関連付けが解除されました 15/06/26 14:39:15 エラー YarnScheduler: 失われましたhslave1 のエグゼキュータ 2: リモート Akka クライアントの関連付けが解除されました [ステージ 0:> (0 + 2) / 2]15/06/26 14:39:33 エラー YarnScheduler: hslave1 でエグゼキュータ 4 が失われました: リモート Akka クライアントの関連付けが解除されました 15/06/26 14:39:33 エラー TaskSetManager: ステージ 0.0 のタスク 0 が 4 回失敗しました。ジョブの中止
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6, hslave1): ExecutorLostFailure (executor 4 lost)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1203)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1192)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1191)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1191)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:693)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1393)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1354)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
私が行った間違いがあれば教えてください。