Verwenden der Airflow Experimental Rest API zum Auslösen einer DAG

Mit der experimentellen API von Airflow können Sie eine DAG über HTTP auslösen. Das ist praktisch, wenn Sie mit einem Cloud-Speicher wie dem Azure Blob Store arbeiten. Denn obwohl Airflow über das Konzept der Sensoren verfügt, können Sie mit einem externen Auslöser die Abfrage nach dem Erscheinen einer Datei vermeiden. In diesem Blog-Beitrag zeige ich, wie wir Azure Functions verwenden, um einen DAG auszulösen, wenn eine Datei in einen Azure Blob Store hochgeladen wird.

Experimentelle API

Mit der experimentellen API können Sie Informationen über DAGs und Aufgaben abrufen, aber auch DAGs auslösen und sogar löschen. In diesem Blogpost werden wir sie verwenden, um eine DAG auszulösen.Die experimentelle API ist standardmäßig ungesichert. Bevor wir fortfahren, sollten wir daher ein definieren, das sie sichert.Es stehen mehrere Optionen zur Verfügung. In diesem Blogpost verwenden wir das Backend , das HTTP Basic Authentication implementiert. Etwas, das Sie nur über HTTPS verwenden sollten.

Die Aktivierung des password_auth Backends ist eine kleine Änderung Ihrer Airflow-Konfigurationsdatei:

[api]
auth_backend = airflow.contrib.auth.backends.password_auth

Als nächstes verwenden Sie die Airflow-Webschnittstelle, um einen neuen Benutzer zu erstellen, der von der Azure-Funktion zum Auslösen der DAG verwendet werden soll.

Azure Funktionen

Azure ermöglicht es Ihnen, kleine Codeschnipsel zu definieren, die von einer ganzen Reihe anderer Auzre-Produkte ausgelöst werden können. Beispiele sind die Auslösung durch eine Nachricht auf einem EventHub oder in diesem Fall durch eine Datei, die in einem Blob Store erscheint. Es gibt eine Reihe verschiedener Sprachen zur Auswahl. In diesem Fall war ich etwas faul und habe mich im BlobTrigger-Assistenten für JavaScript entschieden.

Sie müssen ein Speicherkonto verknüpfen und den Pfad definieren, den Sie überwachen möchten. In meinem Fall habe ich es so konfiguriert, dass my-data/{name} überwacht wird, wobei my-data der Name des Containers im Speicherkonto ist.

Als nächstes wird Ihnen ein kleines Beispiel präsentiert. Ich habe das Beispiel ein wenig erweitert, was zu folgendem Code führte:

var anfordern. = erfordern('Anfrage');

Modul.exportiert = Funktion (Kontext, myBlob) {
    Kontext.Protokoll("JavaScript Blob-Trigger-Funktion verarbeitet Blob n Name:", Kontext.bindingData.Name, "n Blob Größe:", myBlob.Länge, "Bytes");
    Dateiname = Kontext.bindingData.Name;
    Kontext.Protokoll("DAG triggern");

    anfordern.({
        url: "https://AIRFLOW_URL/api/experimental/dags/DAG_NAME/dag_runs",
        Methode: "POST",
        json: {'conf': ' {"Dateiname": "'+Dateiname+'"}'},
        auth: {
            'Benutzer': 'AIRFLOW_USER',
            'passen': AIRFLOW_PASSWORD'.,
            'sofort senden': true
        }
    }).auf('Antwort', Funktion(Antwort) {
        Kontext.Protokoll(Antwort.statusCode)
    });

    Kontext.fertig();
};

Sie müssen die Felder AIRFLOW_URL, DAG_NAME, AIRFLOW_USER und AIRFLOW_PASSWORD anpassen. Das Schöne dabei ist, dass ich Airflow den Dateinamen der neuen Datei übergebe, den ich später in der DAG verwenden kann.

Luftstrom DAG

Den vollständigen Airflow DAG selbst werde ich nicht posten, aber in dem Auszug unten zeige ich, wie man den filename im DAG verwendet.

def copy_blob(vorlagen_dikt, **kwargs):
    wenn vorlagen_dikt['Dateiname']:
        #Machen Sie etwas mit dem Dateinamen
        passieren
    sonst:
        #kein Dateiname angegeben, wahrscheinlich ein manueller Lauf
        passieren

mit DAG(....) als dag:
    copy_blob = PythonOperator(aufgabe_id=copy_blob'.,
                               bereitstellen_kontext=True,
                               python_aufrufbar=copy_blob,
                               vorlagen_dikt={'Dateiname': "{{dag_run.conf['filename']}}"})

Schlusswort

In diesem Blog-Beitrag habe ich gezeigt, wie man eine Azure-Funktion verwendet, um einen Airflow-DAG mit Hilfe der experimentellen API auszulösen. Mir gefällt sehr gut, dass die Abfrage einer Datei entfällt und dass wir Airflow auf diese Weise gut in ein ansonsten cloudbasiertes Setup integrieren können.

Möchten Sie Ihre Kenntnisse über Apache Airflow verbessern?

Wussten Sie, dass wir einen Apache Airflow-Kurs anbieten, in dem Sie die Interna, die Terminologie und die besten Praktiken bei der Arbeit mit Airflow kennenlernen und praktische Erfahrungen beim Schreiben und Verwalten von Datenpipelines sammeln können?