Eleven Labs Blog – Aufbau einer Datenplattform, Feedback (REX)

Der Kontext von REX rund um den Aufbau einer Datenplattform für unseren Kunden

Im Rahmen einer Data Engineer-Mission für einen Kunden der Studio Eleven Labs, bin ich der Abteilung „Data Factory“ beigetreten, um das Nutzerverhalten zu analysieren und zu verstehen. Dies trägt dazu bei, die Hinzufügung von Funktionen und Produkten, die eingeführt werden sollen, besser steuern zu können.

Ein Poc („Proof of Concept“) wurde vom Datenteam implementiert. Es basiert auf einer ELT-Pipeline (Extrahieren, Laden, Transformieren) unter Verwendung der folgenden Technologien: Google Cloud Platform, Talend, dbt und Power BI.

Um den PoC schnell zu testen, wird die Pipeline auf Jenkins ausgeführt. Allerdings bleibt die Abkehr von der Verwendung von Jenkins zur Pipeline-Ausführung nicht ohne Auswirkungen. Jenkins ist für diese Arbeit nicht geeignet: kein erneuter Versuch, Schreiben der Pipeline in Groovy, nur eine Ausführungsumgebung.

Wie können wir die Verarbeitung zuverlässiger machen und den Bereitstellungsprozess industrialisieren?

In diesem Zusammenhang beginnt meine Mission.

Die ELT-Pipeline: Extrahieren, Laden, Transformieren

Bevor ich mit der Arbeit begann, interessierte ich mich für die Funktionsweise der aktuellen Pipeline.

Wie funktioniert diese Pipeline? Welche Schritte sind zu befolgen? Was sind die Anforderungen dieser Pipeline?

In seinem Funktionsprinzip sucht es nach Daten in verschiedenen Quellen, lädt sie in ein Data Warehouse, transformiert und erstellt neue Datenstrukturen, um sie dann anzuzeigen.

Es ist notwendig, den aktuellen Betrieb der Pipeline vollständig zu verstehen, bevor mit Änderungen begonnen wird. Im Folgenden analysieren wir die Funktionsweise und listen die Hauptkomponenten auf.

Extrahieren und Laden von Daten in Google BigQuery

Die erste Phase beginnt mit der Datenextraktion. Die Pipeline stellt eine Verbindung zu verschiedenen Datenquellen her.

In Datenquellen habe ich:

Für die Extraktionsphase und das Laden in Google BigQueryTalend wurde für diese Arbeit eingerichtet. Es handelt sich um ein Tool, mit dem Sie vollständige ETL-Pipelines (Extract, Transform, Load; nicht zu verwechseln mit ELT) erstellen können. Hier wurde es nur für die Extraktions- und Ladephase in Google BigQuery verwendet.

Entwicklung und Änderung erfordern einen dicken Client und eine manuelle Kompilierung der Pull-Pipeline.

Sobald die Daten in BigQuery vorliegen, kann die Transformationsphase beginnen.

Transformation mit dbt

Der zweite Schritt ist die Datentransformation.

Diese Transformation wird durchgeführt von dbt direkt in das Google BigQuery Data Warehouse.

Mit dbt können Sie Datentransformationen und Vorlagen für SQL-Abfragen organisieren. Google BigQuery führt die SQL-Abfragen aus.

Während dieser Transformationsphase werden neue Datenstrukturen erstellt, um das Ergebnis der Berechnungen zu speichern. Diese Datenaggregate werden dann von einem Datenvisualisierungstool angezeigt: Power BI.

Daten mit Power BI anzeigen

Im letzten Schritt dieser Pipeline erfolgt schließlich die Anzeige der Daten.

Das Endziel all dieser Arbeit besteht darin, die Durchführung aller Berechnungen bei der Anzeige der Analyseberichte zu vermeiden. Ohne die vorgelagerte Arbeit der Berechnung und Datenaggregation würde die Darstellung der Grafiken sehr lange dauern.

Diese Pipeline ist funktionsfähig und bei Jenkins bereits vorhanden. Sehen wir uns die Architektur der neuen Datenplattform an.

Architektur der Datenplattform

Im vorherigen Teil haben wir gesehen, wie die ELT-Pipeline in Jenkins funktioniert. Das Ziel besteht darin, dies in eine robustere Plattform zu übertragen, die für diese Art von Arbeit geeignet ist.

Dafür benötigen wir ein Tool, um diese verschiedenen Phasen der Pipeline zu orchestrieren und im Fehlerfall neu zu starten. Apache Airflow ist der perfekte Kandidat. Google bietet eine verwaltete Version an: Google Composer.

Die Voraussetzungen für diese neue Infrastruktur sind wie folgt:

• Verwenden Sie Google Composer
• Nutzen Sie möglichst viele von Google verwaltete Tools, um die Wartung zu erleichtern
• Infrastruktur als Code mit Terraform
• Separate, dedizierte Umgebungen zum Testen
• Der gesamte zum Ausführen einer Aufgabe erforderliche Code befindet sich in einem Docker-Image
• Überwachung und Alarmierung im Fehlerfall

Wir haben daher das folgende Diagramm:

Das Diagramm ist ziemlich dicht, wir werden es aufschlüsseln.

Zunächst einmal gibt es eine Trennlinie zwischen der Dateninfrastruktur und der sogenannten Devops-Infrastruktur, der die Datenbanken gehören. Diese Abgrenzung spiegelt sich im Infrastrukturkodex wider und ermöglicht eine klare Abgrenzung der Verantwortlichkeiten zwischen den Teams.

Daher finden wir im oberen Teil des Diagramms die Datenquellen vom Typ Datenbank, die vom E-Commerce-Team von devops verwaltet werden. Wir werden Anträge auf Zugang zu diesen Quellen stellen.

Im unteren Teil finden wir die gesamte Dateninfrastruktur. Es gibt viele von Google verwaltete Dienste.

Folgende Leistungen können wir auflisten:

• Geheimmanager
• Artefaktregister
• Komponist
• Cloud-Speicher
• Cloud-IAM
• Cloud-Protokollierung
• Cloud-Überwachung
• BigQuery

Und speziell für die Entwicklungsumgebung haben wir:

Die gesamte Installation und Konfiguration der Infrastruktur erfolgt mit Terraform.

Sobald die Architektur entworfen und dem Team kommuniziert wurde, können wir sie umsetzen.

Konditionierung der Arbeitsbelastung

Sobald die Infrastruktur mit Terraform konfiguriert ist, muss noch die zuvor beschriebene ELT-Pipeline bereitgestellt werden. Es gibt zwei Phasen: Extraktion-Laden und Transformation. Der erste Schritt wird von Talend durchgeführt, der zweite von dbt.

Der Composer-Dienst verwendet Kubernetes, um Apache Airflow auszuführen. In ein paar Worten, Apache Airflow ist eine kostenlose Software, mit der Sie Aufgaben ausführen und planen können.

Daher wäre es interessant, unsere Jobs in Kubernetes auszuführen. Dazu benötigen wir ein Docker-Image.

Talend und dbt sind in Docker-Images verpackt. Sie müssen die Docker-Dateien schreiben und die Images erstellen, die im Artifact Registry-Dienst gespeichert werden. Also mit dem Operator KubernetesPodOperator Unterstützt durch Apache Airflow werden Talend- und dbt-Workloads in Kubernetes ausgeführt.

Die Verwendung von Docker-Images erleichtert die Verwendung verschiedener Tools erheblich, die nicht mit der Composer-Umgebung kompatibel wären.

Ich hatte keine besonderen Schwierigkeiten, abgesehen von der Auswahl des Basis-Images für Talend. Er kein Bild mehr offizielle OpenJDK JRE. Ich musste ein Image von einer der Organisationen suchen und auswählen, die ein brauchbares Docker-Image erstellt. Das von der Adoptium-Organisation bereitgestellte grundlegende Docker-Image schien mir das ausgereifteste zu sein: https://hub.docker.com/_/eclipse-temurin/

Kommen wir zum Bau der Pipeline selbst.

Die Pipeline mit einem orientierten azyklischen Graphen

Talend und dbt sind unsere beiden Hauptbausteine. Es bleibt noch, sie in einer Datei zu organisieren: einem DAG. DAG für Gerichteter azyklischer Graph oder Azyklisch orientierter Graph auf Französisch. Vereinfacht gesagt wird das Diagramm in eine Richtung gelesen, es hat einen Anfang und ein Ende und es ist nicht möglich, zum Anfang des Diagramms zurückzukehren.

flowchart LR
    talend[Extraction-Chargement Talend] --> dbt_model[dbt model] --> refresh_power_bi[Mise à jour PowerBI]

Dieses Diagramm wird auf diese Weise in einen Airflow DAG übersetzt.
from airflow import models
from airflow.providers.cncf.kubernetes.operators.pod import (
    KubernetesPodOperator,
)


with models.DAG(...) as dag:
    talend = KubernetesPodOperator(...)
    dbt_model = KubernetesPodOperator(...)
    refresh_power_bi = KubernetesPodOperator(...)

    talend >> dbt_model >> refresh_power_bi
Wir finden im DAG den Operator KubernetesPodOperatorund schließlich die Reihenfolge der Aufgaben, die von Airflow ausgeführt werden.
Die Erstellung der DAG ist an sich nicht komplex.  Um die Funktionsweise von Airflow zu beherrschen, müssen kleine Feinheiten verstanden werden.
Ich zitiere im Folgenden zwei davon: die unterschiedlichen Daten in Airflow und das Ressourcenmanagement.
Diese beiden Punkte sind wichtig, um zu verstehen, wie Airflow funktioniert.
Auslösedatum, Datendatumsintervall
Zusätzlich zum Begriff des Verarbeitungsauslösedatums gibt es Datenintervalldaten.  Airflow löst die Verarbeitung für einen Datendatumsbereich vor dem Auslösedatum und für die Dauer des nächsten Auslösedatums aus.
Betrachten Sie das folgende Beispiel für eine DAG, die mit konfiguriert ist schedule="0 0 * * *".  Der Luftstrom sollte die Verarbeitung jeden Tag um Mitternacht auslösen.
Für den aktuellen Tag, 18. Oktober 2023, 00:00 UTC

das Auslösedatum: „18. Oktober 2023 00:00 UTC“
Beginn der Datenverarbeitung: 17. Oktober 2023, 00:00 UTC
Enddatum der Datenverarbeitung: 17. Oktober 2023, 23:59 Uhr UTC
das nächste Auslösedatum: „19. Oktober 2023 00:00 UTC“

Für mehr Informationen, https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html#data-interval
Dieser Begriff ist in unserem Anwendungsfall nicht wichtig, aber er ist der Fall, wenn die Verarbeitungsvorgänge Daten über einen bestimmten Zeitraum extrahieren müssen.  Dadurch können Sie nur einen Teil der Daten übernehmen und nicht alle.  Es ist auch möglich, Behandlungen über einen bestimmten Zeitraum hinweg zu wiederholen.
Denken Sie daran, dass die Daten in der UTC-Zeitzone liegen!  Wenn Ihr Composer-Cluster um Mitternacht in der Zeitzone Europa/Paris (also 22:00 Uhr UTC) startet der Tag davor), erfolgt eine doppelte Verarbeitung: eine Verarbeitung für das Datendatumsintervall des Vortags und eine weitere für das Datendatumsintervall des Tages, an dem der Composer gestartet wurde.
Ressourcen verwalten
Die Verwaltung von CPU- und Speicherressourcen ist nicht einfach, insbesondere bei unbekannten Sprachen.
Im Allgemeinen gilt: Je mehr Ressourcen die Arbeitslast hat, desto schneller wird sie verarbeitet.  Dies ist bei Talend der Fall.
Mit 1 CPU und 4 GB Arbeitsspeicher dauerte die Ausführung lange.  Durch die Umstellung auf 4 CPUs und 8 GB halbiert sich die Zeit.
In der DAG wird dies folgendermaßen übersetzt:
from kubernetes.client import models as k8s_models
from airflow.providers.cncf.kubernetes.operators.pod import (
    KubernetesPodOperator,
)

talend = KubernetesPodOperator(
    (...)
    container_resources=k8s_models.V1ResourceRequirements(
        requests={
            "cpu": "4",
            "memory": "8G",
        },
    ),
)
Diagramme in Google Monitoring haben mir dabei geholfen, diese Änderung vorzunehmen und die Ressourcennutzung zu überwachen.
Es ist wichtig, so früh wie möglich im Projekt über ein Überwachungs- und Warnsystem zu verfügen, damit Sie die Entwicklung der Ressourcennutzung schnell erkennen und Abhilfe schaffen können.
Derzeit ist diese neue Infrastruktur noch nicht in Produktion, verfügt aber über alle notwendigen Komponenten, um sie in Produktion zu bringen.
In Produktion gehen
All diese Arbeit ist nutzlos, wenn sie nicht in Produktion geht.  Zur Vorbereitung der Produktion habe ich eine identische Kopie aller ursprünglichen BigQuery-Datensätze für die neue Infrastruktur eingerichtet.  Ich habe den Dienst genutzt Google-Datenübertragungen.
Dann habe ich eine Checkliste geschrieben, um sicherzustellen, dass ich nichts vergessen habe.  Ich antizipiere alle Schritte so weit wie möglich.  Es besteht das Risiko eines vollständigen Datenverlusts während der Umstellung.  Diese Liste muss möglichst explizit und direktiv sein.  Man muss sich entfalten können, ohne Fragen zu stellen.
Ich habe mich mit dem Team synchronisiert, um den Go-Live zu planen.
Am großen Tag wird die Liste ausgerollt.  Sobald die Produktion abgeschlossen ist, erfolgt eine aktive Überwachung der Behandlungen.  Das Monitoring-Dashboard wird täglich überprüft.  Sobald ein Fehler auftritt, wird dieser schnellstmöglich behoben und es erfolgt erneut eine aktive Überwachung dieses Fixes.
Und was kommt als nächstes?
Nach dieser Freigabe in die Produktion wird sich an der Infrastruktur nicht viel ändern.  Es werden vor allem Wartungs- und Aktualisierungsarbeiten, insbesondere am Composer-Dienst, durchgeführt.
Einer der Schwachstellen in der Pipeline ist Talend.  Dieses Tool passt sich nicht gut an eine Cloud-Umgebung an.  Das Projekt wäre, eine alternative Lösung zu finden.  Welches Tool wäre für die Datenextraktion geeignet und welches würde komplett von Google verwaltet?
Abschließend mein Feedback
Der Aufbau dieser Datenplattform war ein großes Projekt von uns. Studio Eleven Labs.  Alles wurde von Grund auf neu gebaut.  Ich habe das Problem gut verstanden und konnte so alle Elemente der Funktionsweise der Pipeline identifizieren.  Die Lösung bestand darin, sich an den Betrieb und die Voraussetzungen anzupassen.  Schließlich verlief die Produktion wie geplant.  Durch die Implementierung einer aktiven Überwachung konnte ich Fehler frühzeitig erkennen.  Dies reduziert die Ausfallzeiten der Plattform erheblich.
Für mich war diese Mission sehr abgeschlossen.  Ich war einmal Architekt mit Infrastrukturdesign, Ops mit dem Schreiben von Terraform und einem guten Verständnis der Google Cloud Platform und schließlich Entwickler mit der Redaktion der DAG Airflow.  Ich komme mit noch mehr Erfahrung davon!