Aufschlüsselung der Worker-Task-Ausführung im Apache DolphinScheduler

Hallo Leute, ich bin Cai Shunfeng, leitender Dateningenieur bei WhaleOps und Committer und PMC-Mitglied der Apache DolphinScheduler-Community. Heute erkläre ich, wie die Worker-Aufgabe von Apache DolphinScheduler funktioniert.

Diese Erklärung gliedert sich in drei Abschnitte:

1. Einführung in Apache DolphinScheduler
2. Überblick über das Gesamtdesign von Apache DolphinScheduler
3. Detaillierter Ausführungsprozess der Worker-Aufgaben

Projekteinführung

Apache DolphinScheduler ist ein verteiltes, leicht erweiterbares Open-Source-System zur visuellen Arbeitsablaufplanung, das für Szenarien auf Unternehmensebene geeignet ist.

Es bietet die folgenden Schlüsselfunktionen und bietet durch visuelle Vorgänge eine Lösung zur Datenverarbeitung über den gesamten Lebenszyklus für Arbeitsabläufe und Aufgaben.

Hauptmerkmale

• Einfach zu bedienen

• Visuelle DAG-Operationen: Benutzer können Komponenten per Drag & Drop auf der Seite verschieben, um sie in einem DAG (Directed Acyclic Graph, gerichteter azyklischer Graph) anzuordnen.

• Plugin-System: Enthält Task-Plugins, Datenquellen-Plugins, Alarm-Plugins, Speicher-Plugins, Registry-Center-Plugins, Cron-Job-Plugins usw. Benutzer können Plugins ganz einfach nach Bedarf erweitern, um ihre Geschäftsanforderungen zu erfüllen.

  
  

• Umfangreiche Nutzungsszenarien

• Statische Konfiguration: Umfasst Workflow-Planung, Online- und Offline-Vorgänge, Versionsverwaltung und Backfill-Funktionen.

• Laufzeitoperationen: Bietet Funktionen wie Pause, Stopp, Fortsetzen und Parameterersetzung.

• Abhängigkeitstypen: Unterstützt eine Vielzahl von Abhängigkeitsoptionen und -strategien und passt sich so an mehr Szenarien an.

• Parameterübergabe: Unterstützt Startparameter auf Workflow-Ebene, globale Parameter, lokale Parameter auf Task-Ebene und dynamische Parameterübergabe.

  
  

• Hohe Zuverlässigkeit

• Dezentrales Design: Alle Dienste sind zustandslos und können horizontal skaliert werden, um den Systemdurchsatz zu erhöhen.

• Überlastschutz und Instanz-Fehlertoleranz:

• Überlastungsschutz: Während des Betriebs überwachen Master und Worker ihre eigene CPU- und Speicherauslastung sowie das Aufgabenvolumen. Bei Überlastung pausieren sie die aktuelle Workflow-/Aufgabenverarbeitung und setzen sie nach der Wiederherstellung fort.

• Fehlertoleranz bei Instanzen: Wenn Master-/Workerknoten ausfallen, erkennt das Registrierungscenter, dass der Serviceknoten offline ist, und führt eine Fehlertoleranz für Workflow- oder Taskinstanzen aus. Dadurch wird die Selbstwiederherstellungsfähigkeit des Systems so weit wie möglich sichergestellt.

Gesamtdesign

Projektarchitektur

Als nächstes stellen wir den allgemeinen Designhintergrund vor. Unten finden Sie das auf der offiziellen Website bereitgestellte Designarchitekturdiagramm.

Aus dem Architekturdiagramm können wir erkennen, dass Apache DolphinScheduler aus mehreren Hauptkomponenten besteht:

• API-Komponente: Der API-Dienst verwaltet in erster Linie Metadaten, interagiert über den API-Dienst mit der Benutzeroberfläche oder ruft API-Schnittstellen auf, um Workflow-Aufgaben und verschiedene vom Workflow benötigte Ressourcen zu erstellen.

  
  

• Masterkomponente: Der Master ist der Controller der Workflowinstanzen und ist verantwortlich für die Verarbeitung von Befehlen, deren Konvertierung in Workflowinstanzen, die Durchführung der DAG-Aufteilung, die geordnete Übermittlung von Aufgaben und die Verteilung der Aufgaben an die Worker.

  
  

• Worker-Komponente: Der Worker ist der Ausführende bestimmter Aufgaben. Nach dem Empfang von Aufgaben verarbeitet er diese entsprechend verschiedener Aufgabentypen, interagiert mit dem Master und meldet den Aufgabenstatus. Insbesondere interagiert der Worker-Dienst nicht mit der Datenbank; nur die API-, Master- und Alarmdienste interagieren mit der Datenbank.

  
  

• Alarmdienst: Der Alarmdienst sendet Alarme über verschiedene Alarm-Plugins. Diese Dienste registrieren sich beim Registrierungszentrum, und Master und Worker melden regelmäßig Heartbeats und den aktuellen Status, um sicherzustellen, dass sie Aufgaben normal empfangen können.

Master- und Worker-Interaktionsprozess

Der Interaktionsprozess zwischen Master und Worker läuft wie folgt ab:

• Aufgabenübermittlung: Nachdem der Master die DAG-Aufteilung abgeschlossen hat, übermittelt er Aufgaben an die Datenbank und wählt eine geeignete Arbeitergruppe aus, um die Aufgaben basierend auf verschiedenen Verteilungsstrategien zu verteilen.

  
  

• Aufgabenempfang: Nachdem der Mitarbeiter eine Aufgabe erhalten hat, entscheidet er anhand ihres Zustands, ob er die Aufgabe annimmt. Es wird eine Rückmeldung gegeben, ob die Annahme erfolgreich war oder nicht.

  
  

• Aufgabenausführung: Der Worker verarbeitet die Aufgabe, aktualisiert den Status auf „Wird ausgeführt“ und gibt das Feedback an den Master weiter. Der Master aktualisiert den Aufgabenstatus und die Startzeitinformationen in der Datenbank.

  
  

• Aufgabenabschluss: Nachdem die Aufgabe abgeschlossen ist, sendet der Worker eine Abschlussereignisbenachrichtigung an den Master und der Master gibt eine ACK-Bestätigung zurück. Wenn keine ACK empfangen wird, versucht der Worker es erneut, um sicherzustellen, dass das Aufgabenereignis nicht verloren geht.

Mitarbeiteraufgabe Empfang

Wenn der Mitarbeiter eine Aufgabe erhält, werden die folgenden Vorgänge ausgeführt:

• Füllt die Hostinformationen aus.
• Generiert den Protokollpfad auf dem Arbeitscomputer.
• Generiert einen Worker Task Executor, der zur Ausführung an den Thread-Pool übermittelt wird.

Der Worker prüft, ob er überlastet ist. Wenn ja, lehnt er die Aufgabe ab. Nachdem der Master die Rückmeldung zum Fehler bei der Aufgabenverteilung erhalten hat, wählt er basierend auf der Verteilungsstrategie weiterhin einen anderen Worker für die Aufgabenverteilung aus.

Worker-Ausführungsprozess

Der konkrete Ausführungsprozess von Worker-Tasks umfasst die folgenden Schritte:

1. Aufgabeninitialisierung: Initialisiert die für die Aufgabe erforderliche Umgebung und Abhängigkeiten.
2. Aufgabenausführung: Führt die spezifische Aufgabenlogik aus.
3. Aufgabenabschluss: Nachdem die Aufgabenausführung abgeschlossen ist, werden die Ergebnisse der Aufgabenausführung an den Masterknoten gemeldet.

Als Nächstes werden wir den spezifischen Aufgabenausführungsprozess detailliert beschreiben.

Bevor die Aufgabenausführung beginnt, wird zunächst ein Kontext initialisiert. An diesem Punkt wird die Startzeit der Aufgabe festgelegt. Um die Genauigkeit der Aufgabe sicherzustellen, ist es notwendig, die Zeit zwischen Master und Worker zu synchronisieren, um Zeitabweichungen zu vermeiden.

Anschließend wird der Task-Status auf „Läuft“ gesetzt und an den Master zurückgemeldet, um ihn darüber zu informieren, dass die Ausführung der Task gestartet wurde.

Da die meisten Aufgaben auf dem Linux-Betriebssystem ausgeführt werden, sind Mandanten- und Dateiverarbeitung erforderlich:

• Mandantenverarbeitung: Zunächst wird geprüft, ob der Mandant vorhanden ist. Wenn nicht, wird basierend auf der Konfiguration entschieden, ob der Mandant automatisch erstellt werden soll. Dazu muss der Bereitstellungsbenutzer über Sudo-Berechtigungen verfügen, um während der Aufgabenausführung zum angegebenen Mandanten wechseln zu können.
• Bestimmte Benutzer : Für einige Szenarien ist es nicht erforderlich, den Mandanten zu wechseln, sondern die Aufgabe einfach mit einem bestimmten Benutzer auszuführen. Dies wird auch vom System unterstützt.

• 
  

Nach der Verarbeitung des Mandanten erstellt der Worker das spezifische Ausführungsverzeichnis. Das Stammverzeichnis des Ausführungsverzeichnisses ist konfigurierbar und erfordert eine entsprechende Berechtigung. Standardmäßig sind die Verzeichnisberechtigungen auf 755 eingestellt.

Während der Aufgabenausführung werden möglicherweise verschiedene Ressourcendateien benötigt, z. B. das Abrufen von Dateien aus AWS S3- oder HDFS-Clustern. Das System lädt diese Dateien zur späteren Verwendung in das temporäre Verzeichnis des Workers herunter.

In Apache DolphinScheduler können Parametervariablen ersetzt werden. Die Hauptkategorien umfassen:

• Integrierte Parameter: Betrifft in erster Linie den Ersatz zeit- und datumsbezogener Parameter.
• Benutzerdefinierte Parameter: Vom Benutzer im Workflow oder in der Aufgabe festgelegte Parametervariablen werden ebenfalls entsprechend ersetzt.

Durch die obigen Schritte sind die Ausführungsumgebung und die erforderlichen Ressourcen der Aufgabe bereit und die Ausführung der Aufgabe kann offiziell beginnen.

Verschiedene Arten von Aufgaben

In Apache DolphinScheduler werden verschiedene Aufgabentypen unterstützt, die jeweils für unterschiedliche Szenarien und Anforderungen gelten. Im Folgenden stellen wir mehrere wichtige Aufgabentypen und ihre spezifischen Komponenten vor.

Diese Komponenten werden häufig zum Ausführen von Skriptdateien verwendet und sind für verschiedene Skriptsprachen und Protokolle geeignet:

• Shell: Führt Shell-Skripte aus.
• Python: Führt Python-Skripte aus.
• SQL: Führt SQL-Anweisungen aus.
• Gespeicherte Prozedur: Führt in der Datenbank gespeicherte Prozeduren aus.
• HTTP: Führt HTTP-Anfragen aus.

Die kommerzielle Version (WhaleScheduler) unterstützt auch das Ausführen von Java-Anwendungen durch die Ausführung von JAR-Paketen.

Komponenten logischer Aufgaben

Mit diesen Komponenten wird die logische Steuerung und das Workflow-Management umgesetzt:

• Schalter: Bedingungslose Steuerungsaufgabe.
• Abhängig: Abhängigkeitsaufgabe.
• SubProzess: Unteraufgabe.
• NextLoop (kommerzielle Version): Schleifensteuerungsaufgabe, geeignet für Finanzszenarien.
• Trigger-Komponente: Überwacht, ob Dateien oder Daten vorhanden sind.

Big Data-Komponenten

Diese Komponenten werden hauptsächlich für die Verarbeitung und Analyse großer Datenmengen verwendet:

• SeaTunnel: Entspricht der kommerziellen Version von WhaleTunnel und wird zur Big Data-Integration und -Verarbeitung verwendet.
• AWS EMR: Amazon EMR-Integration.
• HiveCli: Hive-Befehlszeilenaufgabe.
• Spark: Spark-Aufgabe.
• Flink: Flink-Aufgabe.
• DataX: Datensynchronisierungsaufgabe.

Containerkomponenten

Diese Komponenten werden zum Ausführen von Aufgaben in einer Containerumgebung verwendet:

• K8S: Kubernetes-Aufgabe.

Datenqualitätskomponenten

Wird zur Sicherstellung der Datenqualität verwendet:

• DataQuality: Aufgabe zur Überprüfung der Datenqualität.

Interaktive Komponenten

Diese Komponenten werden für die Interaktion mit Data Science- und Machine Learning-Umgebungen verwendet:

• Jupyter: Jupyter-Notebook-Aufgabe.
• Zeppelin: Zeppelin-Notizbuch-Aufgabe.

Machine Learning-Komponenten

Diese Komponenten werden für die Verwaltung und Ausführung von Machine-Learning-Aufgaben verwendet:

• Kubeflow: Kubeflow-Aufgabe.
• MlFlow: MlFlow-Aufgabe.
• Dvc: Aufgabe der Datenversionskontrolle.

Insgesamt unterstützt Apache DolphinScheduler drei bis vier Dutzend Komponenten und deckt Bereiche von der Skriptausführung über die Verarbeitung großer Datenmengen bis hin zum maschinellen Lernen ab. Weitere Informationen und eine ausführliche Dokumentation finden Sie auf der offiziellen Website .

Aufgabentypabstraktion

In Apache DolphinScheduler werden Aufgabentypen in mehrere Verarbeitungsmodi abstrahiert, um verschiedenen Laufzeitumgebungen und Anforderungen gerecht zu werden.

Nachfolgend stellen wir den Abstraktions- und Ausführungsprozess von Aufgabentypen im Detail vor.

Der Worker ist ein auf einem Server bereitgestellter JVM-Dienst. Für einige Skriptkomponenten (wie Shell und Python) und lokal ausgeführte Aufgaben (wie Spark Local) wird zur Ausführung ein separater Prozess gestartet.

An diesem Punkt interagiert der Worker mit diesen Aufgaben über die Prozess-ID (PID).

Unterschiedliche Datenquellen können unterschiedliche Anpassungen erfordern. Für SQL- und Stored-Procedure-Aufgaben haben wir eine abstrahierte Handhabung für unterschiedliche Datenquellen wie MySQL, PostgreSQL, AWS Redshift usw. Diese Abstraktion ermöglicht eine flexible Anpassung und Erweiterung verschiedener Datenbanktypen.

Remote-Aufgaben beziehen sich auf Aufgaben, die auf Remote-Clustern ausgeführt werden, wie etwa AWS EMR, SeaTunnel-Cluster, Kubernetes-Cluster usw. Der Worker führt diese Aufgaben nicht lokal aus, sondern übermittelt sie an die Remote-Cluster und überwacht deren Status und Nachrichten. Dieser Modus eignet sich besonders für Cloud-Umgebungen, in denen Skalierbarkeit erforderlich ist.

Aufgabenausführung

Protokollsammlung

Verschiedene Plug-Ins verwenden unterschiedliche Verarbeitungsmodi, weshalb die Protokollsammlung entsprechend variiert:

• Lokale Prozesse: Protokolle werden durch Überwachen der Prozessausgabe aufgezeichnet.

• Remote-Aufgaben: Protokolle werden gesammelt, indem der Aufgabenstatus und die Ausgabe vom Remote-Cluster (z. B. AWS EMR) regelmäßig überprüft und in den lokalen Aufgabenprotokollen aufgezeichnet werden.

  
  

Parametervariablensubstitution

Das System durchsucht die Taskprotokolle, um Parametervariablen zu identifizieren, die dynamisch ersetzt werden müssen. Beispielsweise kann Task A im DAG einige Ausgabeparameter generieren, die an den nachgelagerten Task B übergeben werden müssen.

Dabei liest das System die Protokolle ein und ersetzt die Parametervariablen nach Bedarf.

Abrufen der Aufgaben-ID

• Lokale Prozesse: Die Prozess-ID (PID) wird abgerufen.
• Remote-Aufgaben: Die ID der Remote-Aufgabe (z. B. AWS EMR-Aufgaben-ID) wird abgerufen.

Das Speichern dieser Task-IDs ermöglicht weitere Datenabfragen und Remote-Task-Operationen. Wenn beispielsweise ein Workflow gestoppt wird, kann die entsprechende Abbruch-API unter Verwendung der Task-ID aufgerufen werden, um die laufende Aufgabe zu beenden.

Fehlertoleranzbehandlung

• Lokale Prozesse: Wenn ein Worker-Knoten ausfällt, bemerkt der lokale Prozess dies nicht und die Aufgabe muss erneut übermittelt werden.
• Remote Tasks: Läuft der Task auf einem Remote-Cluster (z.B. AWS), kann der Status des Tasks anhand der Task-ID überprüft und versucht werden, den Task zu übernehmen. Bei Erfolg ist kein erneutes Absenden des Tasks notwendig, was Zeit spart.

Abschluss der Aufgabenausführung

Nach der Ausführung einer Aufgabe sind mehrere Abschlussaktionen erforderlich:

• Prüfung der Aufgabenerledigung: Das System prüft, ob eine Warnmeldung gesendet werden muss. Wenn beispielsweise bei einer SQL-Aufgabe die Abfrageergebnisse eine Warnmeldung auslösen, interagiert das System über RPC mit dem Warndienst, um die Warnmeldung zu senden.

• Ereignisrückmeldung: Der Worker sendet das Ereignis zur Aufgabenerledigung (Fertigstellungsereignis) zurück an den Master. Der Master aktualisiert den Aufgabenstatus in der Datenbank und fährt mit dem DAG-Statusübergang fort.

• Kontextbereinigung: Der Worker entfernt den Taskkontext, der zu Beginn der Aufgabe erstellt wurde, aus dem Speicher. Außerdem werden die während der Aufgabenausführung generierten Dateipfade bereinigt. Im Debugmodus (Entwicklungsmodus) werden diese Dateien nicht bereinigt, sodass die Fehlerbehebung bei fehlgeschlagenen Aufgaben möglich ist.

  
  

Durch diese Schritte wird der gesamte Ausführungsprozess einer Taskinstanz abgeschlossen.

Beitrag der Gemeinschaft

Wenn Sie an Apache DolphinScheduler interessiert sind und zur Open-Source-Community beitragen möchten, können Sie gerne unsere Beitragsrichtlinien beachten.

Die Community ermutigt zu aktiven Beiträgen, einschließlich, aber nicht beschränkt auf:

• Melden von Problemen, die während der Nutzung aufgetreten sind.
• Einreichen von Dokumentation und Code-PRs.
• Hinzufügen von Unit-Tests (UT).
• Codekommentare hinzufügen.
• Beheben von Fehlern oder Hinzufügen neuer Funktionen.
• Verfassen technischer Artikel oder Teilnahme an Meetups.

Leitfaden für neue Mitwirkende

Neue Mitwirkende können in den GitHub-Problemen der Community nach Problemen suchen, die als good first issue gekennzeichnet sind. Diese Probleme sind im Allgemeinen einfacher und für Benutzer geeignet, die ihren ersten Beitrag leisten.

Zusammenfassend haben wir etwas über das Gesamtdesign von Apache DolphinScheduler und den detaillierten Ausführungsprozess von Worker-Tasks gelernt.

Ich hoffe, dieser Inhalt hilft Ihnen, Apache DolphinScheduler besser zu verstehen und zu verwenden. Wenn Sie Fragen haben, können Sie mich gerne im Kommentarbereich kontaktieren.