Wie man Daten mit der NASA API extrahiert und speichert: Eine Einführung in DAGs mit Apache Airflow

Im Kontext der Automatisierung und Orchestrierung von Workflows spielt Apache Airflow eine Schlüsselrolle, besonders wenn es darum geht, wiederholbare und skalierbare Prozesse zu gestalten. In dieser Anleitung erläutern wir, wie man eine einfache Datenpipeline erstellt, die täglich ein Bild von der NASA API extrahiert, dieses speichert und eine Benachrichtigung über den Abschluss der Aufgabe ausgibt. Die Verwendung von Diagrammen zur Visualisierung der Architektur und der Datenflüsse ist ein wichtiger Schritt, um die geplante Pipeline im Voraus zu planen und eventuelle Probleme zu identifizieren, bevor sie auftreten.

Ein besonders praktisches Hilfsmittel, um den gesamten Prozess visuell darzustellen und mit verschiedenen Beteiligten zu teilen, sind Tools wie Microsoft PowerPoint, Google Slides, Figma, Miro und Lucidchart. Diese Tools ermöglichen es, Diagramme zu erstellen, die nicht nur die Struktur des Systems zeigen, sondern auch eine Versionshistorie bieten und jederzeit zugänglich sind. Auf diese Weise können Stakeholder den Fortschritt überwachen und Feedback geben, was in der Regel Zeit und Frustration spart.

Die NASA API und das Bild des Tages

In diesem Beispiel extrahieren wir täglich ein Bild aus der "Astronomy Picture of the Day"-API (APOD) von NASA. Das Bild wird in einem lokalen Ordner gespeichert und eine Benachrichtigung wird ausgegeben, sobald der Prozess abgeschlossen ist. Das Wichtigste ist, sicherzustellen, dass die API-Anfragen und -Verbindungen ordnungsgemäß funktionieren. Ein erster Schritt ist die Erstellung eines API-Keys bei der NASA, den man über die Webseite (https://api.nasa.gov/) anfordert. Nach der Registrierung erhält man per E-Mail einen API-Key, der sicher aufbewahrt werden sollte, um unbefugten Zugriff zu vermeiden.

Erstellung einer API-Anfrage in Jupyter Notebook

Bevor der DAG in Airflow erstellt wird, empfiehlt es sich, den Python-Code zunächst in einer lokalen Entwicklungsumgebung wie Jupyter Notebook zu testen. Hier geht es darum, sicherzustellen, dass die API korrekt aufgerufen wird, das Bild wie erwartet abgerufen wird und das Speichern auf der Festplatte reibungslos funktioniert. Um diesen Vorgang zu demonstrieren, zeigen wir ein einfaches Codebeispiel, das folgende Schritte umfasst:

1. Zunächst werden die notwendigen Bibliotheken importiert: requests für HTTP-Anfragen, json zum Parsen der API-Antwort und datetime zur Generierung des aktuellen Datums. Der API-Key wird sicher aus einer lokalen Datei (NASA_Keys.py) geladen.

2. Die URL für die Anfrage wird zusammen mit dem API-Key erstellt. Mit der requests-Bibliothek wird eine HTTP GET-Anfrage an die NASA-API gesendet, die eine Antwort im JSON-Format zurückgibt.

3. In der Antwort finden wir das hochauflösende Bild des Tages (unter dem Key hdurl), das dann heruntergeladen und unter einem eindeutigen Namen gespeichert wird, der das aktuelle Datum enthält, um Namenskonflikte zu vermeiden.

4. Schließlich wird das Bild im lokalen Verzeichnis gespeichert, und wir können überprüfen, ob es korrekt abgelegt wurde.

Die Bedeutung der Visualisierung und der Planung

Es ist äußerst hilfreich, im Vorfeld Diagramme zu erstellen, die den Datenfluss und die Architektur des geplanten Systems darstellen. Gerade bei komplexeren Pipelines können zusätzliche Diagramme notwendig werden, um etwa den Ablauf der Datenverarbeitung, die Reihenfolge der Operationen oder die verschiedenen Komponenten und ihre Interaktionen darzustellen. Diese visuellen Darstellungen helfen nicht nur dabei, das Design zu klären, sondern erleichtern auch die Kommunikation mit anderen Beteiligten und die Fehlerbehebung.

Der einfache Ablauf, der in diesem Beispiel gezeigt wird, mag trivial erscheinen, aber in größeren und komplexeren Systemen sind präzise Planung und regelmäßige Überprüfung der einzelnen Schritte unverzichtbar. Insbesondere bei der Integration externer APIs oder der Arbeit mit verschiedenen Datenquellen kann es zu Problemen kommen, die ohne ein klares Design und gründliche Tests schwer zu diagnostizieren sind.

Skalierbarkeit und weitere Optionen

In der Praxis ist es entscheidend, dass eine Pipeline nicht nur die grundlegenden Anforderungen erfüllt, sondern auch skalierbar ist. Eine wichtige Überlegung ist dabei, wie oft und in welchem Umfang die API-Anfragen durchgeführt werden sollen. In unserem Beispiel verwenden wir eine tägliche Anfrage, aber in vielen Szenarien könnte eine häufigere oder weniger häufige Abfrage sinnvoll sein. Zudem könnte die Notwendigkeit bestehen, das Bild zu transformieren oder mit anderen Datenquellen zu kombinieren, bevor es gespeichert wird. Hier bietet Apache Airflow durch seine flexible DAG-Struktur (Directed Acyclic Graph) eine leistungsstarke Möglichkeit, solche Erweiterungen und Anpassungen vorzunehmen.

Wichtige Aspekte der Fehlerbehandlung und Sicherheit

Neben der Funktionalität ist es unerlässlich, Fehlerbehandlung und Sicherheitsaspekte zu berücksichtigen. In diesem Fall sollten API-Fehler oder Netzwerkausfälle im Code abgefangen und sinnvoll behandelt werden, damit der Prozess nicht ohne Vorwarnung scheitert. Zudem muss darauf geachtet werden, dass API-Keys sicher aufbewahrt werden, um Datenlecks oder Missbrauch zu verhindern. Eine gängige Praxis ist es, API-Schlüssel nicht in öffentlichen Repositories oder Code-Dateien zu speichern, sondern in gesicherten Umgebungen wie Umgebungsvariablen oder speziellen Schlüssel-Management-Diensten.

Die Implementierung eines robusten Fehlermanagements, das etwa automatisch Benachrichtigungen bei fehlerhaften API-Anfragen versendet, kann helfen, Probleme schnell zu identifizieren und zu beheben.

Wie man benutzerdefinierte Provider in Airflow erstellt und verwendet

Die Entwicklung benutzerdefinierter Provider für Apache Airflow eröffnet neue Möglichkeiten, externe Systeme nahtlos in Workflows zu integrieren. Dieser Prozess umfasst die Definition von Verbindungen, das Erstellen von Hooks und Operatoren sowie die Implementierung von Sensoren und Triggern, um mit den externen Systemen in einer effizienten und skalierbaren Weise zu kommunizieren.

Um einen benutzerdefinierten Provider für Airflow zu erstellen, müssen mehrere grundlegende Komponenten definiert werden. Zunächst einmal ist es erforderlich, neue Felder für die Verbindungstypen zu erstellen. In diesem Beispiel fügen wir die Felder „Pot Designator“ und „Additions“ zu einem Teekannen-Verbindungstyp hinzu. Diese Felder werden als Schlüssel im „extra“-Dictionary der Verbindungsobjekte gespeichert. Der folgende Code zeigt, wie dies durchgeführt wird:

python
Copy code
@staticmethod
def get_ui_field_behaviour():
    return {
        "hidden_fields": ["password", "login", "schema", "extra"],
        "placeholders": {
            "pot_designator": "1",
            "additions": "sugar",
            "host": "tea-pot",
            "port": "8083",
        },
        "relabeling": {}
    }

Der Code oben beschreibt, wie UI-Elemente geändert werden. Insbesondere werden Felder, die in der Liste „hidden_fields“ enthalten sind, ausgeblendet, und Platzhalter werden in den entsprechenden Feldern angezeigt, um dem Benutzer Beispiele zu bieten. Das „relabeling“-Feld bleibt leer, kann aber verwendet werden, um die Bezeichnung eines bestehenden Feldes zu ändern, ohne den tatsächlichen Attributnamen in der Verbindungsklasse zu beeinflussen.

Der nächste Schritt besteht darin, eine Methode zu definieren, die eine Verbindung aus einer Datenbank abruft und in den Attributen einer Klasse speichert. Dies wird in der Methode „get_conn“ umgesetzt:

python
Copy code
@cached_property
def get_conn(self):
    conn = self.get_connection(self.tea_pot_conn_id)
    self.url = f"{conn.host}:{conn.port}"
    self.pot_designator = conn.extra_dejson.get("pot_designator", None)
    self.additions = conn.extra_dejson.get("additions", None)
    return self

Diese Methode stellt sicher, dass die Verbindung mit den angegebenen Parametern korrekt geladen wird. Im speziellen Fall dieses Beispiels wird die URL für den Teekannen-Server zusammengestellt, und die Felder „pot_designator“ und „additions“ werden, wenn vorhanden, als Klassenattribute gesetzt.

Ein weiterer wichtiger Bestandteil eines Providers ist die Fähigkeit, eine Verbindung zu testen. Die Methode „test_connection“ stellt sicher, dass die Verbindung aktiv ist und gibt den Status zurück:

python
Copy code
def test_connection(self):
    """Test a connection"""
    if self.is_ready():
        return (True, "Alive")
    return (False, "Not Alive")

Diese Methode wird jedes Mal ausgeführt, wenn der „Test“-Button in der Airflow-Verbindungsoberfläche betätigt wird. Sie prüft die „ready“-Endpunkt des Teekannen-Servers und stellt sicher, dass der Service ordnungsgemäß funktioniert.

Darüber hinaus müssen wir den neuen Verbindungstyp in der Provider-Informationsfunktion registrieren, um Airflow mitzuteilen, dass eine neue Verbindung („teapot“) verfügbar ist:

python
Copy code
"connection-types": [
    {
        "connection-type": "teapot",
        "hook-class-name": "airflow_provider_tea_pot.hooks.TeaPotHook"
    }
]

Dies informiert Airflow darüber, dass der Teekannen-Verbindungstyp mit der „TeaPotHook“-Klasse im Hook-Modul unseres Providers verbunden ist. Dadurch wird der Provider im Airflow-UI sichtbar und kann dort entsprechend konfiguriert werden.

Nachdem die grundlegenden Komponenten des Providers erstellt wurden, können wir mit der Entwicklung der Operatoren beginnen. Ein Operator ist eine der zentralen Komponenten in Airflow, die für die Ausführung von Aufgaben verantwortlich ist. In unserem Beispiel entwickeln wir einen einfachen Operator, der eine Teekanne „macht“:

python
Copy code
class MakeTeaOperator(BaseOperator):
    template_fields = ()

    def __init__(self, tea_pot_conn_id, additions=None, pot_designator=None, **kwargs):
        super().__init__(**kwargs)
        self.tea_pot_conn_id = tea_pot_conn_id
        self.pot_designator = pot_designator
        self.additions = additions

    def execute(self, context):
        self.hook = TeaPotHook(tea_pot_conn_id=self.tea_pot_conn_id)
        if self.pot_designator:
            self.hook.pot_designator = self.pot_designator
        if self.additions:
            self.hook.additions = self.additions
        return self.hook.make_tea()

Jeder Operator erfordert eine Initialisierungsmethode und eine Ausführungsmethode. Die Initialisierung speichert die Konfigurationsparameter, während die Ausführungsmethode den Hook verwendet, um die erforderlichen Arbeiten zu erledigen.

Ein weiteres wichtiges Konzept in Airflow ist der Sensor. Ein Sensor ist eine spezielle Art von Operator, der kontinuierlich einen externen Zustand überwacht, bis ein bestimmtes Ereignis eintritt. Im Beispiel unten haben wir einen Sensor entwickelt, der den Wasserstand in einer Teekanne überwacht:

python
Copy code
class WaterLevelTrigger(BaseTrigger):
    def __init__(self, tea_pot_conn_id, minimum_level) -> None:
        self.tea_pot_conn_id = tea_pot_conn_id
        self.minimum_level = minimum_level

    def serialize(self) -> typing.Tuple[str, typing.Dict[str, typing.Any]]:
        return "airflow_provider_tea_pot.triggers.WaterLevelTrigger", {
            "minimum_level": self.minimum_level,
            "tea_pot_conn_id": self.tea_pot_conn_id
        }

    async def run(self):
        hook = TeaPotHook(tea_pot_conn_id=self.tea_pot_conn_id)
        async_get_water_level = sync_to_async(hook.get_water_level)
        while True:
            rv = await async_get_water_level()
            if json.loads(rv).get('level') > self.minimum_level:
                yield TriggerEvent(rv)

Dieser Trigger überprüft kontinuierlich den Wasserstand der Teekanne und löst ein Ereignis aus, sobald der Wasserstand über dem festgelegten Mindestwert liegt. Dabei wird der synchrone „get_water_level“-Aufruf in eine asynchrone Funktion umgewandelt.

Abschließend lässt sich sagen, dass beim Erstellen von benutzerdefinierten Providern in Airflow nicht nur die reibungslose Kommunikation zwischen Airflow und dem externen System gewährleistet werden muss, sondern auch sicherheitsrelevante und performante Aspekte berücksichtigt werden müssen. Besonders wichtig ist es, die Defensivmaßnahmen zu implementieren, wie etwa die Handhabung von Fehlerfällen oder die Sicherstellung, dass asynchrone Aufgaben korrekt verwaltet werden.

Wie Datenorchestrierung mit Apache Airflow die Effizienz in verschiedenen Branchen steigert

Die Datenorchestrierung hat sich in der modernen Softwareentwicklung und Datenverarbeitung als entscheidender Faktor für den Erfolg in vielen Unternehmen etabliert. Vor allem in Zeiten wachsender Datenmengen und zunehmend komplexer werdender Datenstrukturen ist die Notwendigkeit, diese Daten effizient zu verwalten und zu verarbeiten, unverzichtbar. Apache Airflow, als führendes Open-Source-Tool zur Orchestrierung von Datenpipelines, spielt dabei eine zentrale Rolle. In diesem Abschnitt betrachten wir, was unter Datenorchestrierung zu verstehen ist, wie Apache Airflow als Lösung eingesetzt wird und welche Best Practices sich bewährt haben.

Datenorchestrierung ist der Prozess der Koordination, Automatisierung und Überwachung von Datenworkflows. Dieser Prozess stellt sicher, dass Aufgaben in einer bestimmten Reihenfolge ausgeführt werden und relevante Daten zur richtigen Zeit und am richtigen Ort zur Verfügung stehen. Im Wesentlichen geht es darum, verschiedene miteinander verbundene Aufgaben oder Prozesse so zu verwalten, dass sie effizient und fehlerfrei ausgeführt werden. In einer Welt, in der Unternehmen mit riesigen Mengen an Daten aus unterschiedlichsten Quellen konfrontiert sind, ist eine präzise und zuverlässige Orchestrierung von entscheidender Bedeutung.

Apache Airflow ist eines der mächtigsten Werkzeuge, um genau diese Art von Orchestrierung zu gewährleisten. Airflow wurde entwickelt, um Arbeitsabläufe zu definieren, zu planen und deren Ausführung zu überwachen. Dabei ermöglicht das Tool eine klare Strukturierung und visuelle Darstellung von Datenpipelines, die das Management der verschiedenen Prozesse erheblich vereinfacht. Durch seine Flexibilität und Erweiterbarkeit ist Apache Airflow für eine Vielzahl von Industrien und Use Cases geeignet.

In der Praxis wird Apache Airflow zur Automatisierung von ETL/ELT-Prozessen (Extract, Transform, Load) eingesetzt, die für die Verarbeitung und das Laden von Daten unerlässlich sind. Ein typisches Beispiel ist die Nutzung von Airflow in der E-Commerce-Branche, wo das Tool dabei hilft, Daten aus unterschiedlichen Quellen – etwa Bestellungen, Kundeninteraktionen und Lagerbestände – zu extrahieren, zu transformieren und zu laden. Dies ermöglicht eine schnellere und präzisere Analyse und hilft Unternehmen dabei, fundierte Entscheidungen zu treffen und ihre Geschäftsprozesse effizienter zu gestalten.

Im Finanzsektor, insbesondere bei Banken und Fintech-Unternehmen, wird Airflow genutzt, um den Datentransfer und die Datenverarbeitung zu orchestrieren. Dies umfasst unter anderem die Verwaltung von Transaktionsdaten, die Überwachung von Zahlungsströmen und das Risikomanagement. Durch die Automatisierung dieser komplexen Prozesse mit Airflow können Unternehmen nicht nur Fehler minimieren, sondern auch die Geschwindigkeit und Genauigkeit ihrer Datenverarbeitung erheblich steigern.

Ein weiteres Beispiel ist der Einsatz von Apache Airflow in der Gesundheitsbranche, wo das Tool dazu beiträgt, Daten aus verschiedenen Quellen wie Patientenakten, Laborberichten und klinischen Studien zu integrieren und zu analysieren. Die Orchestrierung von Datenpipelines ermöglicht es, schneller auf kritische Informationen zuzugreifen und somit die Qualität der Versorgung zu verbessern.

Ein wesentlicher Vorteil von Apache Airflow liegt in seiner Flexibilität und Skalierbarkeit. Es lässt sich problemlos in unterschiedliche IT-Umgebungen integrieren, sei es auf einer lokalen Maschine oder in der Cloud. Zudem unterstützt Airflow eine Vielzahl von Datenspeicherlösungen und Datenbanken, wie etwa PostgreSQL, MySQL oder Amazon S3. Dies erleichtert die Anpassung an spezifische Unternehmensbedürfnisse und ermöglicht eine nahtlose Integration in bestehende Systeme.

Die Nutzung von Apache Airflow bringt jedoch auch Herausforderungen mit sich. Eine der wichtigsten Anforderungen für den erfolgreichen Einsatz von Airflow ist die richtige Planung und Strukturierung der Workflows. Airflow erfordert eine sorgfältige Definition von Abhängigkeiten und Prioritäten zwischen den einzelnen Tasks, um sicherzustellen, dass diese in der richtigen Reihenfolge ausgeführt werden. Eine fehlerhafte oder schlecht strukturierte Workflow-Definition kann zu Verzögerungen und Ausfällen führen.

Ein weiteres wichtiges Thema ist das Monitoring. Zwar bietet Apache Airflow umfangreiche Funktionen zur Überwachung und Protokollierung von Workflows, jedoch ist es unerlässlich, die richtigen Metriken und Alarme zu konfigurieren, um Probleme frühzeitig zu erkennen und schnell zu beheben. Eine regelmäßige Überprüfung und Optimierung der Workflows ist ebenfalls notwendig, um sicherzustellen, dass die Orchestrierung auch bei wachsenden Datenmengen und zunehmender Komplexität effizient bleibt.

In Bezug auf Best Practices empfiehlt es sich, Airflow in einem Cluster-Setup zu betreiben, um die Skalierbarkeit und Ausfallsicherheit zu maximieren. Eine verteilte Architektur ermöglicht es, mehrere Worker zu betreiben, die die Ausführung von Tasks parallelisieren und so die Gesamtlaufzeit der Pipelines verkürzen. Auch der Einsatz von Docker-Containern für die Ausführung von Tasks kann dazu beitragen, die Flexibilität und Portabilität von Airflow zu erhöhen.

Zusätzlich ist es wichtig, sich mit den verschiedenen Erweiterungen und Integrationen von Airflow vertraut zu machen. Airflow bietet eine Vielzahl von Operatoren, Hooks und Sensoren, die den Zugriff auf unterschiedliche Systeme und Datenquellen ermöglichen. Indem man diese Erweiterungen nutzt, kann man die Funktionalität von Airflow erheblich ausbauen und an die eigenen Bedürfnisse anpassen.

Nicht zuletzt sollten Unternehmen und Entwickler, die Apache Airflow einsetzen, in die Schulung und Weiterbildung ihrer Teams investieren. Die richtige Anwendung von Best Practices und eine fundierte Kenntnis der Airflow-Architektur sind entscheidend, um das volle Potenzial dieses leistungsfähigen Werkzeugs auszuschöpfen. Indem man sich kontinuierlich mit den neuesten Entwicklungen und Funktionen von Airflow auseinandersetzt, bleibt man nicht nur wettbewerbsfähig, sondern kann auch die Effizienz und Zuverlässigkeit der eigenen Datenpipelines langfristig sichern.