Сводка
Я попытался создать ExecutorPlugin . Я реализовал свой класс плагина внутри JAR приложения и установил spark.executor.plugins на имя моего класса плагина. Однако это не работает, и исполнители не запускаются, потому что не могут найти класс плагина (ClassNotFoundException). Единственный способ заставить его работать, это передать JAR приложения как spark.executor.extraClassPath. Почему это так? Я ожидаю, что класс внутри JAR приложения будет виден исполнителю. Я бы предпочел иметь возможность связывать плагины внутри JAR приложения, чтобы не использовать spark.executor.extraClassPath. Спасибо!
Подробности
Вот очень простое воспроизведение ошибки. Класс плагина:
package spark_word_count;
public class SimplePlugin implements org.apache.spark.ExecutorPlugin {
@Override
public void init() {
System.out.println("Hello!");
}
@Override
public void shutdown() {
System.out.println("Goodbye!");
}
}
Я скомпилирую uberjar своего приложения, spark-word-count.jar, и отправляю так, используя скрипт spark-submit:
spark-submit \
--jars file:/mnt/jars/spark-word-count.jar \
--deploy-mode cluster \
--master spark://master:7077 \
--conf spark.executor.plugins=spark_word_count.SimplePlugin \
file:/mnt/jars/spark-word-count.jar \
file:/data/hamlet.txt
Ошибки из журналов Spark
Драйвер
19/10/28 19:29:35 ERROR [org.apache.spark.scheduler.TaskSchedulerImpl] Lost executor 0 on 10.128.99.2: Unable to create executor due to spark_word_count.SimplePlugin
19/10/28 19:29:35 WARN [org.apache.spark.scheduler.TaskSetManager] Lost task 0.0 in stage 0.0 (TID 0, 10.128.99.2, executor 0): ExecutorLostFailure (executor 0 exited caused by one of the running tasks) Reason: Unable to create executor due to spark_word_count.SimplePlugin
Журналы исполнителя
19/10/28 19:29:42 ERROR [org.apache.spark.executor.CoarseGrainedExecutorBackend] Executor self-exiting due to : Unable to create executor due to spark_word_count.SimplePlugin
java.lang.ClassNotFoundException: spark_word_count.SimplePlugin
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.spark.util.Utils$.classForName(Utils.scala:238)
at org.apache.spark.util.Utils$.$anonfun$loadExtensions$1(Utils.scala:2682)
at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:244)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at scala.collection.TraversableLike.flatMap(TraversableLike.scala:244)
at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:241)
at scala.collection.AbstractTraversable.flatMap(Traversable.scala:108)
at org.apache.spark.util.Utils$.loadExtensions(Utils.scala:2680)
at org.apache.spark.executor.Executor.$anonfun$executorPlugins$2(Executor.scala:148)
at org.apache.spark.util.Utils$.withContextClassLoader(Utils.scala:249)
at org.apache.spark.executor.Executor.<init>(Executor.scala:147)
at org.apache.spark.executor.CoarseGrainedExecutorBackend$$anonfun$receive$1.applyOrElse(CoarseGrainedExecutorBackend.scala:83)
at org.apache.spark.rpc.netty.Inbox.$anonfun$process$1(Inbox.scala:117)
at org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:205)
at org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:102)
at org.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:221)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Также я подтвердил, что мой uberjar содержит класс плагина:
$ jar tf spark-word-count.jar | grep SimplePlugin
spark_word_count/SimplePlugin.class
spark_word_count/SimplePlugin.java