Supercharging Airflow & dbt mit Astronomer Cosmos auf Azure Container Instanzen

In den letzten Jahren hat sich dbt zum Standard für Datentransformationen entwickelt. Ein sehr typischer Arbeitsablauf besteht nun darin, dass eine grundlegende Datenextraktion und -eingabe erfolgt, nach der die Transformationen in einem oder mehreren dbt-Projekten durchgeführt werden (ein typischer (Extract-Load-Transform)-Workflow). Im Wesentlichen kümmert sich dbt, wie angekündigt, um die in . Das funktioniert zwar gut, aber ein großer Nachteil ist die Sichtbarkeit und Bedienbarkeit in dem von Ihnen gewählten Datenorchestrator. Der naive Ansatz besteht darin, dbt "nur" als einzelne Aufgabe in Ihrer Pipeline auszuführen. Das funktioniert zwar, macht aber den gesamten Schritt in Ihrem Prozess zu einer undurchsichtigen Aufgabe, was es schwieriger macht, Fehler in Ihren Datenpipelines zu verstehen und zu beheben. Wenn Sie nur eine Aufgabe haben, ist es außerdem unmöglich, Teile Ihres Datenmodells erneut zu versuchen oder erneut auszuführen.

In diesem Blogpost zeigen wir Ihnen, wie Sie diese Undurchsichtigkeit in Transparenz umwandeln können, indem Sie Astronomer Cosmos verwenden, um Ihr dbt-Projekt automatisch in einen Airflow DAG zu rendern, während Sie dbt auf Azure Container Instances ausführen.

Astronom Cosmos stellt sich vor

Astronomer Cosmos ist ein Open-Source-Projekt, das von Astronomer erstellt und gepflegt wird. Sein primäres Ziel ist es, innerhalb von Airflow Transparenz über die Struktur des dbt-Projekts zu schaffen und dabei jegliche manuelle Konfiguration oder Änderungen an Ihrem dbt-Projekt zu vermeiden.

[caption id="" align="alignnone" width="1760"] Bild mit freundlicher Genehmigung von Astronomer: www.astronomer.io/images/cosmos/image1.png[/caption]

Astronomer Cosmos übernimmt Ihr dbt-Projekt und konvertiert es in Aufgaben in einer Airflow DAG. Jedes Modell wird in seine eigene Aufgabe umgewandelt. Alle Tests, die mit diesem Modell verbunden sind, werden in einer separaten Aufgabe ausgeführt, nachdem die Modellaufgabe abgeschlossen ist. Dies geschieht, indem entweder der Quellcode des dbt-Projekts oder das dbt-Projekt analysiert wird. Sie könnten dies natürlich auch selbst tun, indem Sie dbt build --select verwenden, um bestimmte Modelle oder Gruppen von Modellen auszuwählen und für jedes dieser Modelle eine Aufgabe zu erstellen. Dies würde jedoch einige manuelle Arbeit erfordern und auch einen gewissen Wartungsaufwand mit sich bringen, um sicherzustellen, dass der Airflow DAG weiterhin mit dem dbt-Projekt abgeglichen wird.

Wo liegt das Problem?

Astronomer Cosmos benötigt ein paar Dinge, damit es ohne Probleme funktioniert:

1. Ein Airflow-DAG (und die entsprechende Airflow-Einrichtung)
2. Zugriff entweder auf den Quellcode des dbt-Projekts oder auf die dbt-Projekt manifest.json
3. Eine Laufzeit, in der die dbt-Befehle ausgeführt werden.

Dieser letzte Punkt ist das Hauptthema dieses Blogs. In einigen Umgebungen könnte dies so einfach sein wie die Verwendung des Kubernetes-Clusters, auf dem Airflow läuft. Die Aufgaben würden dann in ihren eigenen Pods laufen, die nach Abschluss der jeweiligen Aufgabe wieder entfernt werden. Es gibt jedoch auch Fälle, in denen dies nicht möglich ist:

• Möglicherweise betreiben Sie Airflow in einer eingeschränkten Kubernetes-Umgebung, z.B. mit begrenzten Ressourcen oder Fähigkeiten/Berechtigungen für das Spinnen zusätzlicher Pods
• Wenn Sie Airflow auf einer virtuellen Maschine ausführen, möchten Sie vielleicht keine Abhängigkeiten für eine DAG in der von allen DAGs verwendeten Umgebung installieren.
• Ihre Airflow-Umgebung hat möglicherweise keinen direkten Zugriff auf das Zielsystem (z.B. Databricks, PostgreSQL, BigQuery) aufgrund von Netzwerk- oder Sicherheitseinschränkungen, die in der Umgebung, in der Airflow selbst läuft, gelten.
• Aus Gründen der Compliance ist es Ihnen möglicherweise nicht gestattet, dbt selbst innerhalb einer verwalteten Airflow-Umgebung außerhalb der Kontrolle Ihres Unternehmens auszuführen.
• Möglicherweise haben Sie keinen direkten Zugriff auf eine private Container-Registrierung von Airflow.

Dies sind nur einige Beispiele, bei denen eine Laufzeit für dbt nicht selbstverständlich ist, es gibt sicher noch mehr. Im nächsten Abschnitt erläutern wir einen Ansatz, mit dem wir diese Situationen erfolgreich bewältigt haben.

Lösung: Azure Container-Instanzen

Bei der Arbeit an einem Kundenprojekt sind wir kürzlich auf einen der genannten Fälle gestoßen. In diesem Abschnitt erfahren Sie, was wir getan haben, um die Probleme zu lösen.

Zuvor hatte Astronomer Cosmos Unterstützung für 4 verschiedene Laufzeiten:

1. Docker
2. Kubernetes
3. Lokal
4. Virtualenv

Um die Lücke zu Azure zu schließen, haben wir die Option hinzugefügt, Azure Container Instances (ACI) als Laufzeitumgebung für Astronomer Cosmos zu verwenden(Doku & PR). Mit Azure Container Instances können Sie Container auf Abruf in einer dedizierten Umgebung ausführen. Wenn Sie sie verwenden, sieht unser Setup in etwa so aus:

In diesem Fall werden die Daten, die wir mit dbt transformieren, in einer Azure Database for PostgreSQL-Instanz gespeichert. Wir führen dbt in einem Container in ACI aus, der mit Airflow auf der Grundlage eines in einer privaten Azure Container Registry gespeicherten Images erstellt und zerstört wird.

Die Teilnahme am ACI hat uns einige Vorteile gebracht:

1. Wir verwenden jetzt die nativen Azure-Tools, was einige Integrationen erleichtert.
2. Die ACI-Operatoren von Airflow erstellen und zerstören die ACI-Instanzen automatisch, um sicherzustellen, dass Sie nach Abschluss der Aufträge keine Ressourcen zurücklassen.
3. Da es sich um ein Docker-Image handelt, haben wir die volle Kontrolle über die Laufzeitumgebung (und können sie lokal ausführen). Wir können nun dasselbe Docker-Image in unseren CI/CD-Pipelines verwenden.

Wir fügen nun alle Teile zusammen und zeigen Ihnen, wie sich die Umstellung von einer dbt-Einzelaufgabe mit Azure Container Instances auf eine Astronomer Cosmos-gesteuerte Einrichtung auswirkt, die ebenfalls Azure Container Instances als Laufzeitumgebung verwendet.

Werfen wir einen Blick auf das Beispiel des Jaffle Shops. Wir gehen davon aus, dass Sie ACI ohnehin als Laufzeitumgebung für dbt verwenden möchten. Damit dies funktioniert, benötigen Sie einige Konfigurationen, die sowohl dem Cosmos- als auch dem Nicht-Cosmos-Ansatz gemeinsam sind:

aci_params = {
    "image": "<<< The base image you want to run in ACI >>>",
    "ci_conn_id": "<<< Airflow connection with credentials to create/delete ACI instances in the target resource group >>>",
    "registry_conn_id": "<<< Airflow connection with credentials to pull the image from a container registry >>>",
    "resource_group": "<<< Azure resource group in which the ACI instances are created >>>",
    "name": "astro-aci-{{ ti.task_id.replace('.','-').replace('_','-') }}", # we do some replacement here as _ are not allowed`
    "region": "West Europe",
    "environment_variables": { # Connection configuration for your setup, PostgreSQL is our target in this case.
        "POSTGRES_DATABASE": "{{ conn.aci_db.schema }}",
        "POSTGRES_HOST": "{{ conn.aci_db.host }}",
        "POSTGRES_PASSWORD": "{{ conn.aci_db.password }}",
        "POSTGRES_PORT": "{{ conn.aci_db.port }}",
        "POSTGRES_SCHEMA": "{{ conn.aci_db.schema }}",
        "POSTGRES_USER": "{{ conn.aci_db.login }}",
    },
    "secured_variables": ["POSTGRES_PASSWORD"] # Any variables that you want to pass in but need to obfuscate as they are sensitive.
}

Für den Nicht-Cosmos-Ansatz sieht der Quellcode der DAG etwa so aus::

with DAG(
        dag_id="jaffle_shop_azure_container_instance",
        start_date=datetime(2022, 11, 27),
        schedule=None,
        catchup=False,
) as dag:

    pre_dbt = EmptyOperator(task_id="pre_dbt")
    dbt = AzureContainerInstancesOperator(task_id="dbt", command=["dbt", "build"], **aci_params)

    post_dbt = EmptyOperator(task_id="post_dbt")

    pre_dbt >> dbt >> post_dbt

wodurch Sie eine DAG erhalten, die wie folgt aussieht: Auch das funktioniert, aber es gibt Ihnen wenig bis gar keine Granularität oder Sichtbarkeit, um zu verstehen, welcher Teil fehlgeschlagen ist, noch die Möglichkeit, nur Teile des dbt-Builds erneut auszuführen.

Wenn wir nun zu Astronomer Cosmos wechseln, mit Azure Container Instances als Laufzeitumgebung, erhalten wir etwa Folgendes:

shared_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.AZURE_CONTAINER_INSTANCE,
    dbt_project_path="..."
)

shared_profile_config = ProfileConfig(
    profile_name="jaffle_shop",
    target_name="dev",
    profiles_yml_filepath=DBT_ROOT_PATH / "jaffle_shop" / "profiles.yml"
)

with DAG(
        dag_id="jaffle_shop_azure_container_instance",
        start_date=datetime(2022, 11, 27),
        schedule=None,
        catchup=False,
) as dag:

    pre_dbt = EmptyOperator(task_id="pre_dbt")

    dbt = DbtTaskGroup(
        group_id="dbt",
        project_config=ProjectConfig(
            manifest_path=DBT_ROOT_PATH / "jaffle_shop" / "target" / "manifest.json",
            project_name="jaffle_shop"),
        execution_config=shared_execution_config,
        operator_args=aci_params,
        profile_config=shared_profile_config,
        default_args={"retries": 2},
    )

    post_dbt = EmptyOperator(task_id="post_dbt")

    pre_dbt >> dbt >> post_dbt

die uns diese DAG liefert:

Jede Aufgabe, die Sie sehen, führt dazu, dass ein einzelner dbt-Befehl in einer Azure Container-Instanz ausgeführt wird. Die ACI-Instanz wird automatisch erstellt und nach Beendigung des Befehls auch automatisch wieder entfernt. Am dbt-Projekt selbst waren keinerlei Änderungen erforderlich.

Ein möglicher Nachteil dieser Einrichtung ist der Overhead, der durch die Verwendung von ACI als Laufzeitumgebung für dbt entsteht. Erfahrungsgemäß kann die Startzeit für ACI etwa 1 Minute betragen. Dies kann je nach Ihrem Docker-Image variieren. Dieser Overhead fällt für jede Aufgabe in der dbt/Airflow-DAG an. Das kann unerschwinglich sein, hängt aber von Ihrer Situation ab. Wenn dies der Fall ist, sollten Sie sich nach Konfigurationen mit schnelleren Startzeiten umsehen, wie z.B. Kubernetes (wo die Startzeit der Pods zwar immer noch ein Problem sein kann, aber weniger stark ins Gewicht fällt). Der Betrieb eines Kubernetes-Clusters erfordert jedoch einen wesentlich höheren Wartungsaufwand.

Einpacken

Astronomer Cosmos bietet eine schöne und einfache Möglichkeit, einen besseren Einblick und eine bessere Kontrolle zu erhalten, wenn Sie Ihren dbt-Code in Airflow ausführen. Durch die Hinzufügung von Unterstützung für Azure Container Instances als Laufzeit haben wir die native Integration mit Azure erleichtert, was für Ihren Kontext eine Voraussetzung sein kann.

Foto von Lars Kienle auf Unsplash