CI/CD-Muster mit Workflow Orchestration Manager

Von Bedeutung

Ab dem 1. Januar 2026 können Sie keine neuen Airflow-Instanzen mehr mit dem Workflow Orchestration Manager von ADF erstellen. Wir empfehlen, dass Sie alle Workflow-Orchestration-Manager-Workloads (Apache Airflow in Azure Data Factory) vor dem 31. Dezember 2025 auf Apache Airflow-Jobs in Microsoft Fabric migrieren.

Für weitere Informationen oder Unterstützung während Ihrer Migration zu Apache Airflow in Microsoft Fabric wenden Sie sich bitte an den Microsoft Support.

Workflow Orchestration Manager bietet eine einfache und effiziente Möglichkeit zum Erstellen und Verwalten von Apache Airflow-Umgebungen. Der Dienst ermöglicht es Ihnen, Datenpipelines in großem Maßstab mühelos auszuführen. Es gibt zwei primäre Methoden zum Ausführen von DAGs im Workflow Orchestration Manager. Sie können die DAG-Dateien in Ihren BLOB-Speicher hochladen und mit der Airflow-Umgebung verknüpfen. Alternativ können Sie das Git-Synchronisierungsfeature verwenden, um Ihr Git-Repository automatisch mit der Airflow-Umgebung zu synchronisieren.

Wenn Sie mit Datenpipelines in Airflow arbeiten, müssen Sie ihre DAGs, Plug-Ins und Anforderungsdateien häufig basierend auf Ihren Workflowanforderungen erstellen oder aktualisieren. Entwickler können DAG-Dateien zwar manuell in BLOB-Speicher hochladen oder bearbeiten, aber viele Organisationen bevorzugen die Verwendung eines kontinuierlichen Integrations- und Kontinuierlichen Übermittlungsansatzes (CI/CD) für die Codebereitstellung. Dieser Artikel führt Sie durch die empfohlenen Bereitstellungsmuster, um Ihre Apache Airflow DAGs nahtlos in Workflow Orchestration Manager zu integrieren und bereitzustellen.

Grundlegendes zu CI/CD

Kontinuierliche Integration

Kontinuierliche Integration ist eine Softwareentwicklungspraxis, die häufige und automatisierte Integration von Codeänderungen in ein freigegebenes Repository betont. Es beinhaltet, dass Entwickler regelmäßig ihren Code committen, und bei jedem Commit erstellt eine automatisierte CI-Pipeline den Code, führt Tests aus und führt Validierungsprüfungen durch. Die hauptzieligen Ziele sind es, Integrationsprobleme frühzeitig im Entwicklungsprozess zu erkennen und zu beheben und Entwicklern schnelles Feedback zu geben.

CI stellt sicher, dass die Codebasis in einem ständig testbaren und bereitstellungsfähigen Zustand bleibt. Diese Praxis führt zu verbesserter Codequalität, Zusammenarbeit und der Fähigkeit, Fehler zu erfassen und zu beheben, bevor sie zu erheblichen Problemen werden.

Kontinuierliche Auslieferung

Kontinuierliche Lieferung ist eine Erweiterung von CI, die die Automatisierung um einen Schritt weiter führt. Während bei CI die Automatisierung der Integrations- und Testphasen im Mittelpunkt steht, automatisiert CD die Bereitstellung von Codeänderungen in der Produktionsumgebung oder in anderen Zielumgebungen. Mit dieser Vorgehensweise können Organisationen Softwareupdates schnell und zuverlässig freigeben. Es reduziert Fehler bei der manuellen Bereitstellung und stellt sicher, dass genehmigte Codeänderungen schnell an Benutzer übermittelt werden.

CI/CD-Workflow im Workflow Orchestration Manager

Git-Synchronisierung mit Dev/QA-Integrationslaufzeit

Ordnen Sie Ihre Workflow-Orchestration-Manager-Umgebung mit dem Entwicklungs-/QA-Branch Ihres Git-Repositorys ab.

CI-Pipeline mit Entwicklungs-/QA-Integrationslaufzeit

Wenn eine Pullanforderung (PR) von einem Featurezweig zum Entwicklungszweig hergestellt wird, löst sie eine PR-Pipeline aus. Diese Pipeline wurde entwickelt, um Qualitätsprüfungen in Ihren Featurezweigen effizient durchzuführen und die Codeintegrität und Zuverlässigkeit sicherzustellen. Sie können die folgenden Arten von Prüfungen in die Pipeline einschließen:

• Python-Abhängigkeiten testen: Diese Tests installieren und überprüfen die Richtigkeit von Python-Abhängigkeiten, um sicherzustellen, dass die Abhängigkeiten des Projekts ordnungsgemäß konfiguriert sind.
• Codeanalyse und -linting: Tools für statische Codeanalyse und Linting werden angewendet, um die Codequalität und die Einhaltung von Codierungsstandards zu bewerten.
• DAG-Tests für Den Luftstrom: Diese Tests führen Validierungstests aus, einschließlich Tests für die DAG-Definition und Komponententests, die für Airflow-DAGs entwickelt wurden.
• Komponententests für benutzerdefinierte Airflow-Operatoren, Hooks, Sensoren und Trigger

Wenn eine dieser Prüfungen fehlschlägt, wird die Pipeline beendet. Anschließend müssen Sie die identifizierten Probleme beheben.

Git-Synchronisierung mit Produktionsintegrationslaufzeit

Ordnen Sie die Umgebung Ihres Workflow Orchestration Managers dem Produktionszweig Ihres Git-Repositorys zu.

PR-Pipeline mit Integration Runtime für die Produktion

Eine bewährte Methode besteht darin, eine separate Produktionsumgebung beizubehalten, um zu verhindern, dass jedes Entwicklungsfeature öffentlich zugänglich wird.

Nachdem der Feature-Branch erfolgreich in den Entwicklungszweig zusammengeführt wurde, können Sie einen Pull-Request an den Produktionszweig erstellen, um das neu zusammengeführte Feature öffentlich zu machen. Diese Pull-Request löst die PR-Pipeline aus, die rasche Qualitätsprüfungen für den Entwicklungszweig durchführt. Qualitätsprüfungen stellen sicher, dass alle Features korrekt integriert wurden und es keine Fehler in der Produktionsumgebung gibt.

Vorteile der Verwendung des CI/CD-Workflows im Workflow Orchestration Manager

• Fail-Fast-Ansatz: Ohne Integration des CI/CD-Prozesses werden Sie wahrscheinlich erstmals auf DAG-Fehler aufmerksam, wenn der gerichtete azyklische Graph an GitHub gepusht und mit Workflow Orchestration Manager synchronisiert wird und dabei ein Import Error auftritt. Unterdessen kann ein anderer Entwickler den fehlerhaften Code aus dem Repository unwissentlich abrufen, was potenziell zu Ineffizienzen führt.
• Verbesserung der Codequalität: Wenn Sie grundlegende Prüfungen wie Syntaxüberprüfung, erforderliche Importe und Überprüfungen auf andere bewährte Codierungsmethoden vernachlässigen, erhöhen Sie die Wahrscheinlichkeit, dass Unterpar-Code geliefert wird.

Bereitstellungsmuster im Workflow-Orchestrierungs-Manager

Wir empfehlen zwei Bereitstellungsmuster.

Muster 1: Entwickeln von Datenpipelines direkt im Workflow Orchestration Manager

Sie können Datenpipelines direkt im Workflow Orchestration Manager entwickeln, wenn Sie Muster 1 verwenden.

Voraussetzungen

• Wenn Sie noch kein Azure-Abonnement haben, erstellen Sie ein kostenloses Konto, bevor Sie beginnen. Erstellen oder auswählen Sie eine vorhandene Data Factory-Instanz in der Region, in der die Vorschau des Workflow-Orchestrierungs-Managers unterstützt wird.
• Sie benötigen Zugriff auf ein GitHub-Repository.

Vorteile

• Es ist keine lokale Entwicklungsumgebung erforderlich: Workflow Orchestration Manager übernimmt die zugrunde liegende Infrastruktur, Updates und Wartung, wodurch der Betriebsaufwand für die Verwaltung von Airflow-Clustern reduziert wird. Der Dienst ermöglicht es Ihnen, sich auf das Erstellen und Verwalten von Workflows zu konzentrieren, anstatt die Infrastruktur zu verwalten.
• Skalierbarkeit: Workflow Orchestration Manager bietet die automatische Skalierungsfunktion, um Ressourcen nach Bedarf zu skalieren, um sicherzustellen, dass Ihre Datenpipelines die Erhöhung von Arbeitslasten oder Aktivitätsbrüchen ohne manuelle Eingriffe verarbeiten können.
• Überwachung und Protokollierung: Workflow Orchestration Manager umfasst Diagnoseprotokolle und Überwachung, um die Ausführung Ihrer Workflows nachzuverfolgen, Probleme zu diagnostizieren, Warnungen einzurichten und die Leistung zu optimieren.

Arbeitsablauf

1. Verwenden Sie die Git-Synchronisierungsfunktion.

   In diesem Workflow ist es nicht erforderlich, Ihre eigene lokale Umgebung einzurichten. Stattdessen können Sie mit der Git-Synchronisierungsfunktion beginnen, die von Workflow Orchestration Manager angeboten wird. Mit diesem Feature werden Ihre DAG-Dateien automatisch mit Airflow-Webservern, -Planern und -Mitarbeitern synchronisiert. Jetzt können Sie Ihre Datenpipelines direkt über die Workflow Orchestration Manager-Benutzeroberfläche entwickeln, testen und ausführen.

   Erfahren Sie mehr über die Verwendung des Git-Synchronisierungsfeatures des Workflow Orchestration-Managers.

2. Erstellen Sie einzelne Umgebungen für Feature-Zweige.

   Sie können die Verzweigung aus Ihrem Repository auswählen, um mit dem Workflow Orchestration Manager zu synchronisieren. Mit dieser Funktion können Sie eine einzelne Airflow-Umgebung für jeden Featurezweig erstellen. Auf diese Weise können Entwickler an bestimmten Aufgaben für Datenpipelinen arbeiten.

3. Erstellen Sie eine Pullanforderung.

   Beginnen Sie mit dem Einreichen eines Pull-Requests an die Runtime-Integration der Airflow-Entwicklungsumgebung, nachdem Sie Ihre Features in Ihrer dedizierten Airflow-Umgebung gründlich entwickelt und getestet haben.

Muster 2: Lokales Entwickeln von DAGs und Bereitstellen im Workflow Orchestration Manager

Sie können DAGs lokal entwickeln und in Workflow Orchestration Manager bereitstellen, wenn Sie Muster 2 verwenden.

Voraussetzungen

• Sie benötigen Zugriff auf ein GitHub-Repository.
• Stellen Sie sicher, dass mindestens ein Branch Ihres Code-Repositorys mit dem Workflow Orchestration Manager synchronisiert ist, um die Codeänderungen im Dienst zu sehen.

Vorteile

Sie können den Zugriff auf Azure-Ressourcen nur auf Administratoren beschränken.

Arbeitsablauf

1. Richten Sie eine lokale Umgebung ein.

   Beginnen Sie mit der Einrichtung einer lokalen Entwicklungsumgebung für Apache Airflow auf Ihrem Entwicklungscomputer. In dieser Umgebung können Sie Ihren Airflow-Code entwickeln und testen, einschließlich DAGs und Aufgaben. Mit diesem Ansatz können Sie Pipelines entwickeln, ohne sich auf direkten Zugriff auf Azure-Ressourcen zu verlassen.

2. Verwenden Sie die Git-Synchronisierungsfunktion.

   Synchronisieren Sie den Zweig Ihres GitHub-Repositorys mit dem Workflow Orchestration Manager.

   Erfahren Sie mehr über die Verwendung des Git-Synchronisierungsfeatures des Workflow Orchestration-Managers.

3. Verwenden Sie Workflow Orchestration Manager als Produktionsumgebung.

   Nachdem Sie Datenpipelinen erfolgreich in Ihrem lokalen Setup entwickelt und getestet haben, können Sie eine Pullanforderung an die Verzweigung auslösen, die mit dem Workflow Orchestration Manager synchronisiert wurde. Nachdem die Verzweigung zusammengeführt wurde, verwenden Sie Workflow Orchestration Manager-Features wie automatische Skalierung und Überwachung und Protokollierung auf Produktionsebene.

CI/CD-Beispielpipeline

Weitere Informationen findest du unter:

• Azure DevOps
• GitHub-Aktionen

1. Kopieren Sie den Code eines in Workflow Orchestration Manager-Integration Runtime bereitgestellten DAG mithilfe der Git-Synchronisierungsfunktion.

   from datetime import datetime
   from airflow import DAG
   from airflow.operators.bash import BashOperator
   
   with DAG(
       dag_id="airflow-ci-cd-tutorial",
       start_date=datetime(2023, 8, 15),
       schedule="0 0 * * *",
       tags=["tutorial", "CI/CD"]
   ) as dag:
       # Tasks are represented as operators
       task1 = BashOperator(task_id="task1", bash_command="echo task1")
       task2 = BashOperator(task_id="task2", bash_command="echo task2")
       task3 = BashOperator(task_id="task3", bash_command="echo task3")
       task4 = BashOperator(task_id="task4", bash_command="echo task4")
   
       # Set dependencies between tasks
       task1 >> task2 >> task3 >> task4
   

2. Erstellen Sie eine CI/CD-Pipeline. Sie haben zwei Optionen: Azure DevOps- oder GitHub-Aktionen.

   1. Azure DevOps-Option: Erstellen Sie die Datei azure-devops-ci-cd.yaml , und kopieren Sie den folgenden Code. Die Pipeline löst eine Pullanforderung oder Pushanforderung an den Entwicklungsbranch aus:

      trigger:
      - dev
      
      pr:
      - dev
      
      pool:
        vmImage: ubuntu-latest
      strategy:
        matrix:
          Python3.11:
            python.version: '3.11.5'
      
      steps:
      - task: UsePythonVersion@0
        inputs:
          versionSpec: '$(python.version)'
        displayName: 'Use Python $(python.version)'
      
      - script: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
        displayName: 'Install dependencies'
      
      - script: |
          airflow webserver &
          airflow db init
          airflow scheduler &
          pytest
        displayName: 'Pytest'
      

      Weitere Informationen finden Sie unter Azure Pipelines.

   2. Option "GitHub-Aktionen": Erstellen Eines .github/workflows Verzeichnisses in Ihrem GitHub-Repository.

      1. Erstellen Sie im .github/workflows Verzeichnis eine Datei namens github-actions-ci-cd.yml.

      2. Kopieren Sie den folgenden Code. Die Pipeline wird jedes Mal ausgelöst, wenn ein Pull-Request oder Push-Request an den Entwicklungszweig gesendet wird.

         name: GitHub Actions CI/CD
         
         on:
           pull_request:
             branches:
               - "dev"
           push:
             branches:
               - "dev"
         
         jobs:
           flake8:
             strategy:
               matrix:
                 python-version: [3.11.5]
             runs-on: ubuntu-latest
             steps:
               - name: Check out source repository
                 uses: actions/checkout@v4
               - name: Setup Python
                 uses: actions/setup-python@v4
                 with:
                   python-version: ${{matrix.python-version}}
               - name: flake8 Lint
                 uses: py-actions/flake8@v1
                 with:
                   max-line-length: 120
           tests:
             strategy:
               matrix:
                 python-version: [3.11.5]
             runs-on: ubuntu-latest
             needs: [flake8]
             steps:
               - uses: actions/checkout@v4
               - name: Setup Python
                 uses: actions/setup-python@v4
                 with:
                   python-version: ${{matrix.python-version}}
               - name: Install dependencies
                 run: |
                   python -m pip install --upgrade pip
                   pip install -r requirements.txt
               - name: Pytest
                 run: |
                   airflow webserver &
                   airflow db init
                   airflow scheduler &
                   pytest tests/
         

3. Erstellen Sie im Ordner "Tests" die Tests für Airflow-DAGs. Im Folgenden finden Sie einige Beispiele hierfür:

   1. Mindestens ist es entscheidend, Initialtests durchzuführen, indem Sie import_errors verwenden, um die Integrität und Korrektheit der DAG sicherzustellen. Dieser Test stellt Folgendes sicher:

      • Ihr DAG enthält keine Zyklizität: Zyklizität tritt auf, wenn eine Aufgabe eine Schleife oder zirkuläre Abhängigkeit innerhalb des Workflows bildet und kann zu unerwarteten und unendlichen Ausführungsschleifen führen.

      • Es gibt keine Importfehler: Importfehler können aufgrund von Problemen auftreten, z. B. fehlende Abhängigkeiten, falsche Modulpfade oder Codierungsfehler.  

      • Aufgaben werden ordnungsgemäß definiert: Vergewissern Sie sich, dass die Aufgaben in Ihrer DAG korrekt definiert sind.

        @pytest.fixture()
        
        def dagbag():
            return DagBag(dag_folder="dags")
        
        def test_no_import_errors(dagbag):
            """
            Test Dags to contain no import errors.
            """
            assert not dagbag.import_errors
        

   2. Testen Sie, ob bestimmte DAG-IDs in Ihrem Feature-Branch vorhanden sind, bevor Sie sie in den Entwicklungszweig zusammenführen.

      def test_expected_dags(dagbag):
          """
          Test whether expected dag Ids are present.
          """
          expected_dag_ids = ["airflow-ci-cd-tutorial"]
      
          for dag_id in expected_dag_ids:
              dag = dagbag.get_dag(dag_id)
      
              assert dag is not None
              assert dag_id == dag.dag_id
      

   3. Testen Sie, ob nur genehmigte Tags Ihren DAGs zugeordnet sind. Dieser Test hilft dabei, die Verwendung genehmigter Tags zu erzwingen.

      def test_requires_approved_tag(dagbag):
          """
          Test if DAGS contain one or more tags from list of approved tags only.
          """
          Expected_tags = {"tutorial", "CI/CD"}
          dagIds = dagbag.dag_ids
      
          for id in dagIds:
              dag = dagbag.get_dag(id)
              assert dag.tags
              if Expected_tags:
                  assert not set(dag.tags) - Expected_tags
      

4. Wenn Sie nun eine Pullanforderung an den Entwicklungszweig auslösen, können Sie sehen, dass GitHub-Aktionen die CI-Pipeline auslösen, um alle Tests auszuführen.

Verwandte Inhalte

• Quellcode für airflow.models.dagbag
• Apache Airflow-Komponententests