Pipelineabhängigkeiten in Dataflow verwalten

Viele Apache Beam-Pipelines können mit den Dataflow-Standardlaufzeitumgebungen ausgeführt werden. Bei einigen Anwendungsfällen in der Datenverarbeitung ist es jedoch sinnvoll, zusätzliche Bibliotheken oder Klassen zu verwenden. In diesen Fällen müssen Sie möglicherweise Ihre Pipelineabhängigkeiten verwalten.

In der folgenden Liste finden Sie einige Gründe, warum Sie Ihre Pipelineabhängigkeiten verwalten müssen:

  • Die von der Standardlaufzeitumgebung bereitgestellten Abhängigkeiten reichen für Ihren Anwendungsfall nicht aus.
  • Die Standardabhängigkeiten haben entweder Versionskonflikte oder enthalten Klassen und Bibliotheken, die mit Ihrem Pipelinecode nicht kompatibel sind.
  • Sie müssen Ihre Pipeline an bestimmte Bibliotheksversionen anpinnen.
  • Sie haben eine Python-Pipeline, die mit einer konsistenten Gruppe von Abhängigkeiten ausgeführt werden muss.

Wie Sie Abhängigkeiten verwalten, hängt davon ab, ob Ihrer Pipeline Java, Python oder Go verwendet.

Java

Inkompatible Klassen und Bibliotheken können Probleme mit Java-Abhängigkeiten verursachen. Wenn Ihre Pipeline nutzerspezifischen Code und Einstellungen enthält, darf der Code keine gemischten Bibliotheksversionen enthalten.

Probleme mit Java-Abhängigkeiten

Wenn Ihre Pipeline Probleme mit der Java-Abhängigkeit hat, kann einer der folgenden Fehler auftreten:

  • NoClassDefFoundError: Dieser Fehler tritt auf, wenn eine gesamte Klasse zur Laufzeit nicht verfügbar ist.
  • NoSuchMethodError: Dieser Fehler tritt auf, wenn die Klasse im Klassenpfad eine Version verwendet, die nicht die erforderliche Methode enthält, oder wenn sich die Methodensignatur geändert hat.
  • NoSuchFieldError: Dieser Fehler tritt auf, wenn für die Klasse im Klassenpfad eine Version verwendet wird, der ein zur Laufzeit erforderliches Feld fehlt.
  • FATAL ERROR: Dieser Fehler tritt auf, wenn eine integrierte Abhängigkeit nicht richtig geladen werden kann. Wenn Sie eine Uber-JAR-Datei (Shaded) verwenden, nehmen Sie keine Bibliotheken auf, die Signaturen in derselben JAR-Datei verwenden, z. B. Conscrypt.

Abhängigkeitsverwaltung

Zur Vereinfachung der Abhängigkeitsverwaltung für Java-Pipelines verwendet Apache Beam Bill of Materials (BOM)-Artefakte. Die BOM unterstützt die Tools für die Abhängigkeitsverwaltung bei der Auswahl kompatibler Abhängigkeitskombinationen. Weitere Informationen finden Sie unter Apache Beam SDK für Java-Abhängigkeiten in der Apache Beam-Dokumentation.

Wenn Sie eine BOM mit Ihrer Pipeline verwenden und der Abhängigkeitsliste explizit andere Abhängigkeiten hinzufügen möchten, fügen Sie der Datei pom.xml für das SDK-Artefakt die folgenden Informationen hinzu. Verwenden Sie beam-sdks-java-google-cloud-platform-bom, um die richtige Libraries-BOM zu importieren.

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-google-cloud-platform-bom</artifactId>
      <version>LATEST</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  </dependency>
</dependencies>

Das Artefakt beam-sdks-java-core enthält nur das Core SDK. Sie müssen der Liste der Abhängigkeiten explizit weitere Abhängigkeiten wie E/A und Runner hinzufügen.

Python

Wenn Sie Dataflow-Jobs mit dem Apache Beam Python SDK ausführen, ist die Abhängigkeitsverwaltung in den folgenden Szenarien nützlich:

  • Ihre Pipeline verwendet öffentliche Pakete aus dem Python-Paketindex (PyPI) und Sie möchten diese Pakete remote verfügbar machen.
  • Sie möchten eine reproduzierbare Umgebung erstellen.
  • Um die Startzeit zu verkürzen, sollten Sie Abhängigkeitsinstallationen auf Workern zur Laufzeit vermeiden.

Python-Pipeline-Abhängigkeiten definieren

Obwohl Sie ein einziges Python-Skript oder -Notebook verwenden können, um eine Apache Beam-Pipeline zu schreiben, wird Software in der Python-Umgebung häufig als Pakete verteilt. Wenn sich Ihr Pipelinecode über mehrere Dateien erstreckt, sollten Sie die Pipeline-Dateien als Python-Paket gruppieren, um die Wartung Ihrer Pipeline zu erleichtern.

  • Definieren Sie die Abhängigkeiten der Pipeline in der setup.py-Datei Ihres Pakets.
  • Stellen Sie das Paket mit der Pipeline-Option --setup_file für die Worker bereit.

Wenn die Remote-Worker gestartet werden, installieren sie Ihr Paket. Ein Beispiel finden Sie unter Juliset im Apache Beam GitHub.

So strukturieren Sie Ihre Pipeline als Python-Paket:

  1. Erstellen Sie eine setup.py-Datei für Ihr Projekt. Fügen Sie in der Datei setup.py das Argument install_requires ein, um die Mindestanzahl von Abhängigkeiten für Ihre Pipeline anzugeben. Das folgende Beispiel zeigt eine einfache setup.py-Datei.

    import setuptools

setuptools.setup(
  name='PACKAGE_NAME',
  version='PACKAGE_VERSION',
  install_requires=[],
  packages=setuptools.find_packages(),
)

  2. Fügen Sie die Datei setup.py, die zentrale Workflowdatei und ein Verzeichnis mit den restlichen Dateien dem Stammverzeichnis Ihres Projekts hinzu. Diese Dateigruppierung ist das Python-Paket für Ihre Pipeline. Die Dateistruktur sieht so aus wie das folgende Beispiel:

    root_dir/
  package_name/
    __init__.py
    my_pipeline_launcher.py
    my_custom_transforms.py
    ...other files...
  setup.py
  main.py

  3. Wenn Sie Ihre Pipeline ausführen möchten, installieren Sie das Paket in der Umgebung für die Einreichung. Verwenden Sie die Pipelineoption --setup_file, um das Paket auf den Workern bereitzustellen. Beispiel:

    python -m pip install -e .
python main.py --runner DataflowRunner --setup_file ./setup.py  <...other options...>

Diese Schritte vereinfachen die Wartung des Pipelinecodes, insbesondere wenn der Code an Größe und Komplexität zunimmt. Weitere Möglichkeiten zum Angeben von Abhängigkeiten finden Sie in der Apache Beam-Dokumentation unter Python-Pipelineabhängigkeiten verwalten.

Benutzerdefinierte Container verwenden, um die Laufzeitumgebung zu steuern

Um eine Pipeline mit dem Apache Beam Python SDK auszuführen, benötigen Dataflow-Worker eine Python-Umgebung mit einem Interpreter, dem Apache Beam SDK und den Pipelineabhängigkeiten. Docker-Container-Images bieten die erforderliche Umgebung für die Ausführung Ihres Pipelinecodes.

Mit jeder Version des Apache Beam SDK werden Standard-Container-Images veröffentlicht, die die Apache Beam SDK-Abhängigkeiten enthalten. Weitere Informationen finden Sie unter Apache Beam SDK für Python-Abhängigkeiten in der Apache Beam-Dokumentation.

Wenn Ihre Pipeline eine Abhängigkeit benötigt, die nicht im Standard-Container-Image enthalten ist, muss diese Abhängigkeit zur Laufzeit installiert werden. Die Installation von Paketen zur Laufzeit kann folgende Folgen haben:

  • Die Startzeit des Workers verlängert sich aufgrund der Auflösung, des Herunterladens und der Installation von Abhängigkeiten.
  • Für die Ausführung der Pipeline ist eine Internetverbindung erforderlich.
  • Nichtdeterminismus tritt aufgrund von Software-Releases in Abhängigkeiten auf.

Um diese Probleme zu vermeiden, stellen Sie die Laufzeitumgebung in einem benutzerdefinierten Docker-Container-Image bereit. Die Verwendung eines benutzerdefinierten Docker-Container-Images, in dem die Pipeline-Abhängigkeiten vorinstalliert sind, bietet folgende Vorteile:

  • So wird sichergestellt, dass die Pipeline-Laufzeitumgebung jedes Mal, wenn Sie Ihren Dataflow-Job starten, dieselben Abhängigkeiten hat.
  • Damit können Sie die Laufzeitumgebung Ihrer Pipeline steuern.
  • Verhindert potenziell zeitaufwendige Abhängigkeitsauflösung beim Start.

Beachten Sie die folgenden Hinweise, wenn Sie benutzerdefinierte Container-Images verwenden:

  • Verwenden Sie das Tag :latest nicht mit Ihren benutzerdefinierten Bildern. Taggen Sie Ihre Builds mit einem Datum, einer Version oder einer eindeutigen Kennung. Mit diesem Schritt können Sie bei Bedarf zu einer bekannten funktionierenden Konfiguration zurückkehren.
  • Verwenden Sie eine Startumgebung, die mit Ihrem Container-Image kompatibel ist. Weitere Informationen zur Verwendung benutzerdefinierter Container finden Sie unter Container-Image erstellen.

Weitere Informationen zum Vorinstallieren von Python-Abhängigkeiten finden Sie unter Python-Abhängigkeiten vorinstallieren.

Startumgebung mit Dataflow-Vorlagen steuern

Wenn für Ihre Pipeline zusätzliche Abhängigkeiten erforderlich sind, müssen Sie sie möglicherweise sowohl in der Laufzeitumgebung als auch in der Startumgebung installieren. In der Launch-Umgebung wird die Produktionsversion der Pipeline ausgeführt. Da die Startumgebung mit der Laufzeitumgebung kompatibel sein muss, sollten Sie in beiden Umgebungen dieselben Versionen von Abhängigkeiten verwenden.

Verwenden Sie flexible Dataflow-Vorlagen, um eine containerisierte, reproduzierbare Startumgebung zu erhalten. Weitere Informationen finden Sie unter Flex-Vorlagen zum Verpacken einer Dataflow-Pipeline für die Bereitstellung verwenden. Beachten Sie bei der Verwendung flexibler Vorlagen die folgenden Faktoren:

Dadurch wird Ihre Startumgebung sowohl reproduzierbar als auch mit Ihrer Laufzeitumgebung kompatibel.

Ein Beispiel für diesen Ansatz finden Sie in der Anleitung Flex-Vorlage für eine Pipeline mit Abhängigkeiten und einem benutzerdefinierten Container in GitHub.

Weitere Informationen finden Sie unter Startumgebung mit der Laufzeitumgebung kompatibel machen und Von der Pipeline verwendete Abhängigkeiten steuern in der Apache Beam-Dokumentation.

Go

Wenn Sie Dataflow-Jobs mit dem Apache Beam Go SDK ausführen, werden Go-Module zum Verwalten von Abhängigkeiten verwendet. Die folgende Datei enthält die Standardkompilierungs- und Laufzeitabhängigkeiten, die von Ihrer Pipeline verwendet werden:

https://raw.githubusercontent.com/apache/beam/vVERSION_NUMBER/sdks/go.sum

Ersetzen Sie VERSION_NUMBER durch die von Ihnen verwendete SDK-Version.

Informationen zum Verwalten von Abhängigkeiten für Ihre Go-Pipeline finden Sie in der Go-Dokumentation unter Abhängigkeiten verwalten.