Beheben von Abhängigkeitsproblemen in Apache Spark

Die häufigsten Probleme beim Erstellen und Bereitstellen von Spark-Anwendungen sind:

• java.lang.ClassNotFoundException.
• object x is not a member of package y Kompilierungsfehler.
• java.lang.NoSuchMethodError

Wie können diese gelöst werden?

Der Klassenpfad von Apache Spark wird dynamisch erstellt (um Benutzercode pro Anwendung aufzunehmen), was ihn für solche Probleme anfällig macht. Die Antwort von @ user7337271 ist richtig, aber je nach dem gibt es einige weitere Bedenken Cluster-Manager (“master”) Sie verwenden.

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

1. Treiber: das ist dein Anwendung erstellen a SparkSession (oder SparkContext) und eine Verbindung zu einem Cluster-Manager herstellen, um die eigentliche Arbeit auszuführen
2. Cluster-Manager: dient als “Einstiegspunkt” zum Cluster und ist für die Zuordnung zuständig Vollstrecker für jede Anwendung. In Spark werden verschiedene Typen unterstützt: Standalone, YARN und Mesos, die wir im Folgenden beschreiben werden.
3. Vollstrecker: Dies sind die Prozesse auf den Clusterknoten, die die eigentliche Arbeit ausführen (Ausführen von Spark Aufgaben)

Die Beziehung zwischen diesen wird in diesem Diagramm von Apache Spark beschrieben Übersicht über den Cluster-Modus:

Jetzt – 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 Sparks Bibliotheken. Sie sollten in existieren ALLE drei Komponenten, da sie den Klebstoff enthalten, mit dem Spark die Kommunikation zwischen ihnen durchführen kann. Übrigens – Spark-Autoren haben eine Designentscheidung getroffen, Code für ALLE Komponenten in ALLE Komponenten aufzunehmen (z. B. Code, der nur in Executor ausgeführt werden soll, auch in Treiber aufzunehmen), um dies zu vereinfachen – also Sparks „Fat Jar“ (in Versionen bis 1.6 ) oder “archive” (in 2.0, Details unten) enthalten den notwendigen Code für alle Komponenten und sollten in allen verfügbar sein.

2. Nur-Fahrer-Code Dies ist Benutzercode, der nichts enthält, was auf Executors verwendet werden sollte, dh Code, der in keiner Transformation auf dem RDD / DataFrame / Dataset verwendet wird. Dieser muss nicht unbedingt vom verteilten Benutzercode getrennt werden, kann es aber.

3. Verteilter 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 in diesen JAR(s) enthalten sein.

Jetzt, wo wir das klargestellt haben, wie Lassen sich die Klassen in jeder Komponente korrekt laden, und welchen Regeln sollten sie folgen?

1. Spark-Code: Wie in früheren Antworten angegeben, müssen Sie dasselbe verwenden Skala und Funke Versionen in allen Komponenten.

   1,1 Zoll Eigenständige -Modus gibt es eine „bereits vorhandene“ Spark-Installation, mit der sich Anwendungen (Treiber) verbinden können. Das bedeutet, dass alle Treiber müssen dieselbe Spark-Version verwenden läuft auf Master und Executoren.

   1,2 Zoll GARN / Mesos, kann jede Anwendung eine andere Spark-Version verwenden, aber alle Komponenten derselben Anwendung müssen dieselbe Version verwenden. Das heißt, wenn Sie Version X zum Kompilieren und Packen Ihrer Treiberanwendung verwendet haben, sollten Sie beim Starten der SparkSession dieselbe Version bereitstellen (z spark.yarn.archive oder spark.yarn.jars Parameter bei Verwendung von YARN). Die von Ihnen bereitgestellten Gläser/Archive sollten alle Spark-Abhängigkeiten enthalten (einschließlich transitiver Abhängigkeiten) und wird vom Cluster-Manager an jeden Ausführenden gesendet, wenn die Anwendung gestartet wird.

2. Fahrercode: Das liegt ganz bei Ihnen – Treibercode kann als Bündel von Gläsern oder als “fettes Glas” versendet werden, solange es alle Spark-Abhängigkeiten + den gesamten Benutzercode enthält

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

Zusammenfassenhier ist ein vorgeschlagener Ansatz zum Erstellen und Bereitstellen einer Spark-Anwendung (in diesem Fall mit YARN):

• Erstellen Sie eine Bibliothek mit Ihrem verteilten Code, packen Sie sie sowohl als “normales” JAR (mit einer .pom-Datei, die ihre Abhängigkeiten beschreibt) als auch als “Fat-Jar” (mit all ihren transitiven Abhängigkeiten).
• Erstellen Sie eine Treiberanwendung mit Kompilierungsabhängigkeiten von Ihrer verteilten Codebibliothek und von Apache Spark (mit einer bestimmten Version).
• Packen Sie die Treiberanwendung in ein Fat-Jar, das für den Treiber bereitgestellt werden soll
• Übergeben Sie die richtige Version Ihres verteilten Codes als Wert von spark.jars Parameter beim Starten der SparkSession
• Übergeben Sie den Speicherort einer Archivdatei (z. B. gzip), die alle darunter liegenden Gläser enthält lib/ Ordner der heruntergeladenen Spark-Binärdateien als Wert von spark.yarn.archive

Beim Erstellen und Bereitstellen von Spark-Anwendungen erfordern alle Abhängigkeiten kompatible Versionen.

• Scala-Version. Alle Pakete müssen dieselbe Hauptversion (2.10, 2.11, 2.12) von Scala verwenden.

  Betrachten Sie Folgendes (falsch) build.sbt:

  name := "Simple Project"
  
  version := "1.0"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" % "spark-core_2.11" % "2.0.1",
     "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
     "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
  )
  

  Wir gebrauchen spark-streaming für Scala 2.10, während die restlichen Pakete für Scala 2.11 sind. EIN gültig Datei sein könnte

  name := "Simple Project"
  
  version := "1.0"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" % "spark-core_2.11" % "2.0.1",
     "org.apache.spark" % "spark-streaming_2.11" % "2.0.1",
     "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
  )
  

  Es ist jedoch besser, die Version global anzugeben und zu verwenden %% (was die Scala-Version für Sie anhängt):

  name := "Simple Project"
  
  version := "1.0"
  
  scalaVersion := "2.11.7"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" %% "spark-core" % "2.0.1",
     "org.apache.spark" %% "spark-streaming" % "2.0.1",
     "org.apache.bahir" %% "spark-streaming-twitter" % "2.0.1"
  )
  

Ähnlich in Maven:

    <project>
      <groupId>com.example</groupId>
      <artifactId>simple-project</artifactId>
      <modelVersion>4.0.0</modelVersion>
      <name>Simple Project</name>
      <packaging>jar</packaging>
      <version>1.0</version>
      <properties>
        <spark.version>2.0.1</spark.version>
      </properties> 
      <dependencies>
        <dependency> <!-- Spark dependency -->
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency> 
        <dependency>
          <groupId>org.apache.bahir</groupId>
          <artifactId>spark-streaming-twitter_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
      </dependencies>
    </project>

• Spark-Version Alle Pakete müssen dieselbe Hauptversion von Spark verwenden (1.6, 2.0, 2.1, …).

  Betrachten Sie Folgendes (falsch) build.sbt:

  name := "Simple Project"
  
  version := "1.0"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" % "spark-core_2.11" % "1.6.1",
     "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
     "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
  )
  

  Wir gebrauchen spark-core 1.6, während die restlichen Komponenten in Spark 2.0 enthalten sind. EIN gültig Datei sein könnte

  name := "Simple Project"
  
  version := "1.0"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" % "spark-core_2.11" % "2.0.1",
     "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
     "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
  )
  

  aber es ist besser zu eine Variable verwenden
  (immer noch falsch):

  name := "Simple Project"
  
  version := "1.0"
  
  val sparkVersion = "2.0.1"
  
  libraryDependencies ++= Seq(
     "org.apache.spark" % "spark-core_2.11" % sparkVersion,
     "org.apache.spark" % "spark-streaming_2.10" % sparkVersion,
     "org.apache.bahir" % "spark-streaming-twitter_2.11" % sparkVersion
  )
  

Ähnlich in Maven:

    <project>
      <groupId>com.example</groupId>
      <artifactId>simple-project</artifactId>
      <modelVersion>4.0.0</modelVersion>
      <name>Simple Project</name>
      <packaging>jar</packaging>
      <version>1.0</version>
      <properties>
        <spark.version>2.0.1</spark.version>
        <scala.version>2.11</scala.version>
      </properties> 
      <dependencies>
        <dependency> <!-- Spark dependency -->
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency> 
        <dependency>
          <groupId>org.apache.bahir</groupId>
          <artifactId>spark-streaming-twitter_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
      </dependencies>
    </project>

• Die in Spark-Abhängigkeiten verwendete Spark-Version muss mit der Spark-Version der Spark-Installation übereinstimmen. Zum Beispiel Wenn Sie 1.6.1 auf dem Cluster verwenden, müssen Sie 1.6.1 zum Erstellen von Gläsern verwenden. Nicht übereinstimmende Nebenversionen werden nicht immer akzeptiert.

• Die Scala-Version, die zum Erstellen von JAR verwendet wird, muss mit der Scala-Version übereinstimmen, die zum Erstellen von bereitgestelltem Spark verwendet wird. Standardmäßig (herunterladbare Binärdateien und Standard-Builds):

  • Spark 1.x -> Scala 2.10
  • Spark 2.x -> Scala 2.11

• Zusätzliche Pakete sollten auf den Worker-Knoten zugänglich sein, wenn sie im Fat-Jar enthalten sind. Es gibt eine Reihe von Optionen, darunter:

  • --jars Argument für spark-submit – Lokal zu verteilen jar Dateien.
  • --packages Argument für spark-submit – um Abhängigkeiten aus dem Maven-Repository abzurufen.

  Bei der Einreichung im Cluster-Knoten sollten Sie die Bewerbung einschließen jar in --jars.

Zusätzlich zu der sehr ausführlichen Antwort, die bereits von user7337271 gegeben wurde, können Sie, wenn das Problem auf fehlende externe Abhängigkeiten zurückzuführen ist, ein Glas mit Ihren Abhängigkeiten mit zB erstellen Maven-Assembly-Plugin

Stellen Sie in diesem Fall sicher, dass alle Core-Spark-Abhängigkeiten in Ihrem Build-System als „bereitgestellt“ gekennzeichnet sind, und stellen Sie, wie bereits erwähnt, sicher, dass sie mit Ihrer Runtime-Spark-Version korrelieren.

Abhängigkeitsklassen Ihrer Anwendung sind in der anzugeben Anwendungsglas Option Ihres Startbefehls.

Weitere Einzelheiten finden Sie unter Spark-Dokumentation

Entnommen aus der Dokumentation:

application-jar: Pfad zu einem gebündelten JAR, das Ihre Anwendung und alle Abhängigkeiten enthält. Die URL muss innerhalb Ihres Clusters global sichtbar sein, z. B. ein hdfs://-Pfad oder ein file://-Pfad, der auf allen Knoten vorhanden ist

Ich denke, dieses Problem muss ein Assembly-Plugin lösen. Du musst ein fettes Glas bauen. Zum Beispiel in sbt:

• Datei hinzufügen $PROJECT_ROOT/project/assembly.sbt mit Code addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.0")
• bauen.sbtadded some librariesBibliotheksabhängigkeiten ++= Seq(“com.some.company” %% “some-lib” % “1.0.0”)`
• Geben Sie in der sbt-Konsole “assembly” ein und stellen Sie das Assembly-Jar bereit

Wenn Sie weitere Informationen benötigen, gehen Sie zu https://github.com/sbt/sbt-assembly

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

Ich habe die folgende 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 ein dickes Glas meiner Anwendung mit dem sbt-Assembly-Plugin erstellt, aber wenn es mit spark-submit ausgeführt wird, schlägt es mit dem 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 Fettglas existiert:

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

Ich bin mir nicht sicher, was ich hier übersehe, irgendwelche Hinweise?