Beheben von Abhängigkeitsproblemen in Apache Spark

Der Klassenpfad von Apache Spark wird dynamisch erstellt (um den anwendungsspezifischen Benutzercode zu berücksichtigen), wodurch er für solche Probleme anfällig wird. Die Antwort von @ user7337271 ist richtig, aber es gibt noch einige Bedenken, abhängig vom Cluster-Manager ("master") du benutzt.

Erstens besteht eine Spark -Anwendung aus diesen Komponenten (jede ist eine separate JVM und enthält daher möglicherweise unterschiedliche Klassen in ihrem Klassenpfad):

1. Treiber : das ist Ihr Anwendung, die ein SparkSession (oder SparkContext) und erstellt Herstellen einer Verbindung zu einem Cluster-Manager, um die eigentliche Arbeit auszuführen
2. Cluster-Manager : Dient als "Einstiegspunkt" für den Cluster, der für die Zuweisung von Ausführenden für jede Anwendung zuständig ist. In Spark werden verschiedene Typen unterstützt: Standalone, YARN und Mesos, die im Folgenden beschrieben werden.
3. Executors : Dies sind die Prozesse auf den Clusterknoten, die die eigentliche Arbeit ausführen (Ausführen von Spark Aufgaben )

Die Beziehung zwischen diesen ist in diesem Diagramm aus Apache Spark's Cluster-Modus-Übersicht beschrieben:

Nun - welche Klassen sollten sich in jeder dieser Komponenten befinden?

Dies kann durch das folgende Diagramm beantwortet werden:

Analysieren wir das langsam:

1. Spark-Code sind Spark-Bibliotheken. Sie sollten in [~ # ~] allen [~ # ~] drei Komponenten vorhanden sein, da sie den Klebstoff enthalten, mit dem Spark die Kommunikation zwischen ihnen durchgeführt wird. Übrigens - Die Autoren von Spark haben eine Designentscheidung getroffen, Code für ALLE Komponenten in ALLE Komponenten aufzunehmen (z. B. Code, der nur in Executor im Treiber ausgeführt werden soll), um dies zu vereinfachen Versionen bis 1.6) oder "archive" (in 2.0, Details unten) enthalten den erforderlichen Code für alle Komponenten und sollten in allen von ihnen verfügbar sein.

2. Nur-Treiber-Code Hierbei handelt es sich um Benutzercode, der nichts enthält, was auf Executors verwendet werden sollte, dh Code, der in keiner Transformation auf der RDD/DataFrame/Dataset. Dies muss nicht unbedingt vom verteilten Benutzercode getrennt werden, kann es aber sein.

3. Distributed Code Dies ist Benutzercode, der mit Treibercode kompiliert wird, aber auch auf den Executors ausgeführt werden muss - alles, was die eigentlichen Transformationen verwenden, muss enthalten sein dieses (diese) Glas (e).

Nachdem wir das klargestellt haben, wie bringen wir die Klassen dazu, in jede Komponente richtig geladen zu werden, und welche Regeln sollten sie befolgen?

1. Spark Code : Wie in den vorherigen Antworten angegeben, müssen Sie das gleiche Scala verwenden. und Spark Versionen in allen Komponenten.

   1.1 Im Standalone-Modus gibt es eine "bereits vorhandene" Spark Installation, mit der sich Anwendungen (Treiber) verbinden können. Das bedeutet, dass alle Treiber dieselbe Spark - Version verwenden müssen, die auf dem Master und den Executoren ausgeführt wird.

   1.2 In YARN/Mesos kann jede Anwendung eine andere Spark - Version verwenden, aber alle Komponenten derselben Anwendung müssen dieselbe verwenden. Das bedeutet, wenn Sie Ihre Treiberanwendung mit Version X kompiliert und gepackt haben, sollten Sie beim Starten der SparkSession dieselbe Version bereitstellen (z. B. über die Parameter spark.yarn.archive Oder spark.yarn.jars Bei Verwendung von YARN). Die von Ihnen bereitgestellten JARS/Archive sollten alle Spark - Abhängigkeiten ( einschließlich der transitiven Abhängigkeiten enthalten. Sie werden vom Cluster-Manager an alle ausgeliefert Executor, wenn die Anwendung gestartet wird.

2. Treibercode : Das liegt ganz bei Ihnen. Der Treibercode kann als Bund Gläser oder als "Fettglas" geliefert werden, solange er alle Spark Abhängigkeiten + aller Benutzercode

3. Verteilter Code : Dieser Code muss nicht nur auf dem Treiber vorhanden sein, sondern auch an die Ausführenden gesendet werden (ebenfalls mit all seinen transitiven Abhängigkeiten). Dies geschieht mit dem Parameter spark.jars.

Um zusammenzufassen, hier ein empfohlener Ansatz zum Erstellen und Bereitstellen einer Spark - Anwendung (in diesem Fall mit YARN):

• Erstellen Sie eine Bibliothek mit Ihrem verteilten Code, und verpacken Sie sie als "reguläre" JAR-Datei (mit einer POM-Datei, in der die Abhängigkeiten beschrieben werden) und als "Fat JAR-Datei" (mit allen darin enthaltenen transitiven Abhängigkeiten).
• Erstellen Sie eine Treiberanwendung mit Kompilierungsabhängigkeiten für Ihre verteilte Codebibliothek und für Apache Spark (mit einer bestimmten Version).
• Packen Sie die Treiberanwendung in ein fettes Glas, das für den Treiber bereitgestellt werden soll
• Übergeben Sie die richtige Version Ihres verteilten Codes als Wert des Parameters spark.jars, Wenn Sie SparkSession starten.
• Übergeben Sie den Speicherort einer Archivdatei (z. B. gzip) mit allen Gläsern im Ordner lib/ Der heruntergeladenen Spark - Binärdateien als Wert von spark.yarn.archive.

Zusätzlich zu der sehr umfangreichen Antwort, die bereits von user7337271 gegeben wurde, können Sie, wenn das Problem auf fehlende externe Abhängigkeiten zurückzuführen ist, eine Dose mit Ihren Abhängigkeiten mit z. maven Assembly Plugin

Markieren Sie in diesem Fall alle Kernfunkenabhängigkeiten in Ihrem Build-System als "bereitgestellt" und stellen Sie, wie bereits erwähnt, sicher, dass sie mit Ihrer Laufzeitfunkenversion korrelieren.

Abhängigkeitsklassen Ihrer Anwendung müssen in der Option application-jar Ihres Startbefehls angegeben werden. 

Weitere Details finden Sie in der Spark-Dokumentation

Aus der Dokumentation entnommen:

application-jar: Pfad zu einer gebündelten jar einschließlich Ihrer Anwendung und alle Abhängigkeiten. Die URL muss in Ihrem .__ global sichtbar sein. Cluster, zum Beispiel ein hdfs: // Pfad oder eine Datei: // Pfad, der .__ ist. auf allen Knoten vorhanden

Fügen Sie alle JAR-Dateien aus spark-2.4.0-bin-hadoop2.7\spark-2.4.0-bin-hadoop2.7\jars im Projekt hinzu. Der spark-2.4.0-bin-hadoop2.7 kann von https://spark.Apache.org/downloads.html heruntergeladen werden.

Ich habe folgendes build.sbt 

lazy val root = (project in file(".")).
  settings(
    name := "spark-samples",
    version := "1.0",
    scalaVersion := "2.11.12",
    mainClass in Compile := Some("StreamingExample")        
  )

libraryDependencies ++= Seq(
  "org.Apache.spark" %% "spark-core" % "2.4.0",
  "org.Apache.spark" %% "spark-streaming" % "2.4.0",
  "org.Apache.spark" %% "spark-sql" % "2.4.0",
  "com.couchbase.client" %% "spark-connector" % "2.2.0" 
)

// META-INF discarding
assemblyMergeStrategy in Assembly := {
       case PathList("META-INF", xs @ _*) => MergeStrategy.discard
       case x => MergeStrategy.first
   }

Ich habe mit sbt Assembly Plugin ein fettes Glas meiner Appliction erstellt, aber bei der Ausführung von spark-submit schlägt der Fehler fehl: 

Java.lang.NoClassDefFoundError: rx/Completable$OnSubscribe
    at com.couchbase.spark.connection.CouchbaseConnection.streamClient(CouchbaseConnection.scala:154)

Ich kann sehen, dass die Klasse in meinem fetten Glas existiert: 

jar tf target/scala-2.11/spark-samples-Assembly-1.0.jar | grep 'Completable$OnSubscribe'
rx/Completable$OnSubscribe.class

nicht sicher, was ich hier vermisse, irgendwelche Hinweise? 

Ich denke, dieses Problem muss ein Assembly-Plugin lösen. Sie müssen ein fettes Glas bauen. Zum Beispiel in sbt:

• datei $PROJECT_ROOT/project/Assembly.sbt mit Code addSbtPlugin("com.eed3si9n" % "sbt-Assembly" % "0.14.0") hinzufügen
• to build.sbtadded some librarieslibraryDependencies ++ = Seq ("com.some.company" %% "some-lib"% "1.0.0") ` 
• geben Sie in der sbt-Konsole "Assembly" ein, und stellen Sie das Assembly-Glas bereit 

Weitere Informationen finden Sie unter https://github.com/sbt/sbt-Assembly