Cluster di accensione autonomo. Impossibile inviare il lavoro a livello di programmazione -> java.io.InvalidClassException

Question

Spark fellows, sono abbastanza nuovo per Spark, ecco perché spero davvero il tuo aiuto.

Sto provando a programmare il lavoro abbastanza semplice su spark cluster dal mio laptop. Nonostante funzioni, quando lo invio con ./spark-submit, genera un'eccezione quando provo a farlo a livello di codice.

Ambiente: - Spark - 1 nodo principale e 2 nodi di lavoro (modalità standalone). Spark non è stato compilato, ma i binari sono stati scaricati. Versione Spark - 1.0.2 - java version "1.7.0_45" - Il jar dell'applicazione si trova ovunque (sul client e sui nodi worker nello stesso posto); - Il file README.md viene copiato anche su ogni nodo;

L'applicazione che sto provando a fare funzionare:

val logFile = "/user/vagrant/README.md" 

val conf = new SparkConf() 
conf.setMaster("spark://192.168.33.50:7077") 
conf.setAppName("Simple App") 
conf.setJars(List("file:///user/vagrant/spark-1.0.2-bin-hadoop1/bin/hello-apache-spark_2.10-1.0.0-SNAPSHOT.jar")) 
conf.setSparkHome("/user/vagrant/spark-1.0.2-bin-hadoop1") 

val sc = new SparkContext(conf) 

val logData = sc.textFile(logFile, 2).cache() 

...

Quindi il problema è che questa applicazione funziona su cluster di successo, quando lo faccio:

./spark-submit --class com.paycasso.SimpleApp --master spark://192.168.33.50:7077 --deploy-mode client file:///home/vagrant/spark-1.0.2-bin-hadoop1/bin/hello-apache-spark_2.10-1.0.0-SNAPSHOT.jar 

Ma non funziona, quando provo a fare lo stesso programmaticamente chiamando sbt run

Ecco l'stacktrace, che salgo sul nodo master:

14/09/04 15:09:44 ERROR Remoting: org.apache.spark.deploy.ApplicationDescription; local class incompatible: stream classdesc serialVersionUID = -6451051318873184044, local class serialVersionUID = 583745679236071411 
java.io.InvalidClassException: org.apache.spark.deploy.ApplicationDescription; local class incompatible: stream classdesc serialVersionUID = -6451051318873184044, local class serialVersionUID = 583745679236071411 
    at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:617) 
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1622) 
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517) 
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771) 
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) 
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990) 
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915) 
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798) 
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350) 
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370) 
    at akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136) 
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57) 
    at akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136) 
    at akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104) 
    at scala.util.Try$.apply(Try.scala:161) 
    at akka.serialization.Serialization.deserialize(Serialization.scala:98) 
    at akka.remote.serialization.MessageContainerSerializer.fromBinary(MessageContainerSerializer.scala:58) 
    at akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104) 
    at scala.util.Try$.apply(Try.scala:161) 
    at akka.serialization.Serialization.deserialize(Serialization.scala:98) 
    at akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23) 
    at akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:55) 
    at akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:55) 
    at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:73) 
    at akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:764) 
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498) 
    at akka.actor.ActorCell.invoke(ActorCell.scala:456) 
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237) 
    at akka.dispatch.Mailbox.run(Mailbox.scala:219) 
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386) 
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) 
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) 

Quale potrebbe essere la soluzione a questo? Grazie in anticipo.

Answer 1

Dopo aver perso un sacco di tempo, ho trovato il problema. Nonostante non abbia usato hadoop/hdfs nella mia applicazione, il client hadoop è importante. Il problema era nella versione hadoop-client, era diversa dalla versione di hadoop, per cui la scintilla era stata creata. La versione 1.2.1 di hadoop di Spark, ma nella mia applicazione era 2.4.

Quando ho cambiato la versione del client hadoop in 1.2.1 nella mia app, sono in grado di eseguire codice spark sul cluster.

Answer 2

Invece di usare la libreria normale Spark core, è possibile utilizzare tale dipendenza (definito nel file build.sbt):!

resolvers += "Talend" at "https://talend-update.talend.com/nexus/content/repositories/libraries/" 
libraryDependencies += "org.talend.libraries" % "spark-assembly-1.6.0-hadoop2.6.0" % "6.0.0" 

/\ Coloro scintilla assemblaggio -... Hadoop ... le librairie possono essere abbastanza grandi (e quindi non compatibili con un push Git).

List of Spark/Hadoop Libraries Hosted by Talend

Answer 3

Se si utilizza una scintilla precompilati. (Nessuna installazione di sbt o di maven), assicurarsi che tutti i nodi worker utilizzino la stessa versione di spark. Ho affrontato lo stesso problema in cui uno dei nodi worker utilizzava una versione diversa della scintilla. mantenere le stesse versioni su tutti i nodi risolto il problema per me. Una macchina utilizzava spark-2.0.0 - bin-hadoop2.7 invece di spark-2.0.0-preview-bin-hadoop2.7