Teilen über

Entwickeln von Lakeflow Spark Declarative Pipelines

Das Entwickeln und Testen von Pipelinecode unterscheidet sich von anderen Apache Spark-Workloads. Dieser Artikel enthält eine Übersicht über unterstützte Funktionen, bewährte Methoden und Überlegungen bei der Entwicklung von Pipelinecode. Weitere Empfehlungen und bewährte Methoden finden Sie unter Anwenden bewährter Methoden für die Softwareentwicklung und DevOps auf Pipelines.

Hinweis

Sie müssen einer Pipelinekonfiguration Quellcode hinzufügen, um Code zu überprüfen oder ein Update auszuführen. Siehe Konfigurieren von Pipelines.

Welche Dateien sind für Pipelinequellcode gültig?

Pipelinecode kann Python oder SQL sein. Sie können eine Mischung aus Python- und SQL-Quellcodedateien haben, die eine einzelne Pipeline sichern, aber jede Datei kann nur eine Sprache enthalten. Siehe Entwickeln von Pipelinecode mit Python und Entwickeln von Lakeflow Spark Declarative Pipelines-Code mit SQL.

Quelldateien für Pipelines werden in Ihrem Arbeitsbereich gespeichert. Arbeitsbereichsdateien stellen Python- oder SQL-Skripts dar, die im Lakeflow-Pipelines-Editor erstellt wurden. Sie können die Dateien auch lokal in Ihrer bevorzugten IDE bearbeiten und mit dem Arbeitsbereich synchronisieren. Informationen zu Dateien im Arbeitsbereich finden Sie unter Was sind Arbeitsbereichsdateien?. Informationen zum Bearbeiten mit dem Lakeflow Pipelines Editor finden Sie unter Entwickeln und Debuggen von ETL-Pipelines mit dem Lakeflow Pipelines Editor. Informationen zum Erstellen von Code in einer lokalen IDE finden Sie unter Entwickeln von Pipelinecode in Ihrer lokalen Entwicklungsumgebung.

Wenn Sie Python-Code als Module oder Bibliotheken entwickeln, müssen Sie den Code installieren und importieren und dann Methoden aus einer Python-Datei aufrufen, die als Quellcode konfiguriert ist. Siehe Verwalten von Python-Abhängigkeiten für Pipelines.

Hinweis

Wenn Sie beliebige SQL-Befehle in einem Python-Notizbuch verwenden müssen, können Sie das Syntaxmuster spark.sql("<QUERY>") verwenden, um SQL als Python-Code auszuführen.

Unity Catalog-Funktionen ermöglichen es Ihnen, beliebige benutzerdefinierte Python-Funktionen für die Verwendung in SQL zu registrieren. Siehe benutzerdefinierte Funktionen (USER-Defined Functions, UDFs) im Unity-Katalog.

Übersicht über die Features der Pipelineentwicklung

Pipelines erweitern und nutzen viele Azure Databricks-Entwicklungsfeatures und stellen neue Features und Konzepte vor. Die folgende Tabelle enthält eine kurze Übersicht über Konzepte und Features, die die Entwicklung von Pipelinecode unterstützen:

Merkmal Description
Entwicklungsmodus Wenn Pipelines interaktiv ausgeführt werden (durch Auswahl der Aktualisierung über den Lakeflow Pipelines Editor), wird der Entwicklungsmodus verwendet. Neue Pipelines werden mit deaktiviertem Entwicklungsmodus ausgeführt, wenn sie automatisch über einen Zeitplan oder einen automatisierten Trigger ausgeführt werden. Siehe Entwicklungsmodus.
Probelauf Bei einem Trockenlaufupdate wird die Richtigkeit des Pipelinequellcodes überprüft, ohne dass eine Aktualisierung für tabellen ausgeführt wird. Siehe Überprüfen einer Pipeline auf Fehler, ohne darauf zu warten, dass Tabellen aktualisiert werden.
Lakeflow-Pipeline-Editor Python- und SQL-Dateien, die als Quellcode für Pipelines konfiguriert sind, bieten interaktive Optionen zum Überprüfen von Code und Ausführen von Updates. Siehe Entwickeln und Debuggen von ETL-Pipelines mit dem Lakeflow Pipelines-Editor.
Die Parameter Nutzen Sie Parameter in Quellcode- und Pipelinekonfigurationen, um Tests und Erweiterbarkeit zu vereinfachen. Siehe Verwendung von Parametern mit Pipelines.
Databricks-Ressourcenpakete Databricks Asset Bundles ermöglichen es Ihnen, Pipelinekonfigurationen und Quellcode zwischen Arbeitsbereichen zu verschieben. Siehe Konvertieren einer Pipeline in ein Databricks Asset Bundle-Projekt.

Erstellen von Beispieldatensätzen für Entwicklung und Tests

Databricks empfiehlt das Erstellen von Entwicklungs- und Testdatensätzen zum Testen der Pipelinelogik mit erwarteten Daten und potenziell fehlerhaften oder beschädigten Datensätzen. Es gibt mehrere Möglichkeiten zum Erstellen von Datasets, die für die Entwicklung und tests nützlich sein können, einschließlich der folgenden:

  • Wählen Sie eine Teilmenge von Daten aus einem Produktions-Dataset aus.
  • Verwenden Sie anonymisierte oder künstlich generierte Daten für Quellen, die PII enthalten. Ein Lernprogramm, das die faker Bibliothek zum Generieren von Daten für Tests verwendet, finden Sie unter Lernprogramm: Erstellen einer ETL-Pipeline mithilfe der Änderungsdatenerfassung.
  • Erstellen Sie Testdaten mit gut definierten Ergebnissen basierend auf der nachgeschalteten Transformationslogik.
  • Bereiten Sie sich auf potenzielle Datenbeschädigungen, falsch formatierte Datensätze und Änderungen an vorgelagerten Daten vor, indem Sie Datensätze erstellen, die die Vorgaben des Datenschemas nicht erfüllen.

Wenn Sie beispielsweise über eine Datei verfügen, die ein Dataset mit dem folgenden Code definiert:

CREATE OR REFRESH STREAMING TABLE input_data
AS SELECT * FROM STREAM read_files(
  "/production/data",
  format => "json")

Sie können ein Beispiel-Dataset erstellen, das eine Teilmenge der Datensätze enthält, indem Sie eine Abfrage wie die folgende verwenden:

CREATE OR REFRESH MATERIALIZED VIEW input_data AS
SELECT "2021/09/04" AS date, 22.4 as sensor_reading UNION ALL
SELECT "2021/09/05" AS date, 21.5 as sensor_reading

Im folgenden Beispiel wird das Filtern veröffentlichter Daten veranschaulicht, um eine Teilmenge der Produktionsdaten für die Entwicklung oder Tests zu erstellen:

CREATE OR REFRESH MATERIALIZED VIEW input_data AS SELECT * FROM prod.input_data WHERE date > current_date() - INTERVAL 1 DAY

Um diese verschiedenen Datasets zu verwenden, erstellen Sie mehrere Pipelines mit dem Quellcode, der die Transformationslogik implementiert. Jede Pipeline kann Daten aus dem input_data Dataset lesen, ist jedoch so konfiguriert, dass sie die Datei enthält, die das Dataset erstellt, das für die Umgebung spezifisch ist.

Wie verarbeiten Pipeline-Datasets Daten?

In der folgenden Tabelle wird beschrieben, wie materialisierte Sichten, Streamingtabellen und Sichten Daten verarbeiten:

Datasettyp Wie werden Datensätze über definierte Abfragen verarbeitet?
Streamingtabelle Jeder Datensatz wird genau einmal verarbeitet. Dies setzt eine reine Anfügequelle voraus.
Materialisierte Ansicht Datensätze werden nach Bedarf verarbeitet, um genaue Ergebnisse für den aktuellen Datenzustand zurückzugeben. Materialisierte Ansichten sollten für Datenverarbeitungsaufgaben wie Transformationen, Aggregationen oder vor dem Berechnen langsamer Abfragen und häufig verwendeter Berechnungen verwendet werden. Ergebnisse werden zwischen Updates zwischengespeichert.
Ansicht Datensätze werden jedes Mal verarbeitet, wenn die Ansicht abgefragt wird. Verwenden Sie Ansichten für Zwischentransformationen und Datenqualitätsprüfungen, die nicht für öffentliche Datasets veröffentlicht werden sollen.

Deklarieren Sie Ihre ersten Datensätze in Pipelines

Pipelines führen neue Syntax für Python und SQL ein. Informationen zu den Grundlagen der Pipelinesyntax finden Sie unter Entwickeln von Pipelinecode mit Python und Entwickeln von Lakeflow Spark Declarative Pipelines-Code mit SQL.

Hinweis

Pipelines trennen Datasetdefinitionen von der Aktualisierungsverarbeitung, und die Pipelinequelle ist nicht für die interaktive Ausführung vorgesehen.

Wie konfigurieren Sie Pipelines?

Die Einstellungen für eine Pipeline sind in zwei allgemeine Kategorien unterteilt:

  1. Konfigurationen, die eine Sammlung von Dateien (als Quellcode bezeichnet) definieren, die Pipelinesyntax zum Deklarieren von Datasets verwenden.
  2. Konfigurationen zur Steuerung der Pipelineinfrastruktur, der Abhängigkeitsverwaltung, der Verarbeitung von Updates und der Speicherung von Tabellen im Arbeitsbereich.

Die meisten Konfigurationen sind optional, aber einige erfordern sorgfältige Aufmerksamkeit, insbesondere beim Konfigurieren von Produktionspipelines. Dazu gehören folgende:

  • Um Daten außerhalb der Pipeline verfügbar zu machen, müssen Sie ein Zielschema deklarieren, das im Hive-Metaspeicher oder in einem Zielkatalog und Zielschema veröffentlicht werden soll, um es im Unity-Katalog zu veröffentlichen.
  • Datenzugriffsberechtigungen werden über den cluster konfiguriert, der für die Ausführung verwendet wird. Stellen Sie sicher, dass Ihr Cluster über entsprechende Berechtigungen für Datenquellen und den Zielspeicherort verfügt, sofern angegeben.

Ausführliche Informationen zur Verwendung von Python und SQL zum Schreiben von Quellcode für Pipelines finden Sie unter Pipeline SQL-Sprachreferenz und Lakeflow Spark Declarative Pipelines Python-Sprachreferenz.

Weitere Informationen zu Pipelineeinstellungen und Konfigurationen finden Sie unter Konfigurieren von Pipelines.

Bereiten Sie Ihre erste Pipeline vor und lösen Sie Updates aus

Um Daten mit SDP zu verarbeiten, konfigurieren Sie eine Pipeline. Nachdem eine Pipeline konfiguriert wurde, können Sie eine Aktualisierung auslösen, um Ergebnisse für jedes Dataset in Ihrer Pipeline zu berechnen. Informationen zu den ersten Schritten mit Pipelines finden Sie im Lernprogramm: Erstellen einer ETL-Pipeline mithilfe der Änderungsdatenerfassung.

Was ist ein Pipelineupdate?

Pipelines stellen Infrastruktur bereit und berechnen den Datenstatus neu, wenn Sie ein Update starten. Ein Update führt die folgenden Aktionen aus:

  • Startet einen Cluster mit der richtigen Konfiguration.
  • Ermittelt alle definierten Tabellen und Ansichten und sucht nach Analysefehlern wie ungültigen Spaltennamen, fehlenden Abhängigkeiten und Syntaxfehlern.
  • Erstellt oder aktualisiert Tabellen und Sichten mit den neuesten verfügbaren Daten.

Pipelines können kontinuierlich oder nach einem Zeitplan ausgeführt werden, abhängig von den Kosten- und Latenzanforderungen Ihres Anwendungsfälles. Siehe Ausführen eines Pipelineupdates.

Aufnehmen von Daten mit Pipelines

Pipelines unterstützen alle Datenquellen, die in Azure Databricks verfügbar sind.

Databricks empfiehlt die Verwendung von Streaming-Tabellen für die meisten Datenaufnahmeanwendungsfälle. Für Dateien, die im Cloudobjektspeicher ankommen, empfiehlt Databricks das automatische Laden. Sie können Daten direkt mit einer Pipeline aus den meisten Nachrichtenbussen einlesen.

Weitere Informationen zum Konfigurieren des Zugriffs auf Cloudspeicher finden Sie unter Cloudspeicherkonfiguration.

Bei Formaten, die vom automatischen Laden nicht unterstützt werden, können Sie Python oder SQL verwenden, um ein beliebiges Format abzufragen, das von Apache Spark unterstützt wird. Siehe Laden von Daten in Pipelines.

Überwachung und Durchsetzung der Datenqualität

Sie können die Erwartungen verwenden, um Datenqualitätssteuerelemente für den Inhalt eines Datasets anzugeben. Im Gegensatz zu einer CHECK Einschränkung in einer herkömmlichen Datenbank, die das Hinzufügen von Datensätzen verhindert, die die Einschränkung nicht bestehen, bieten die Erwartungen Flexibilität beim Verarbeiten von Daten, die datenqualitätsanforderungen nicht bestehen. Mit dieser Flexibilität können Sie Daten verarbeiten und speichern, von denen Sie erwarten, dass sie unübersichtlich sind, und Daten, die strenge Qualitätsanforderungen erfüllen müssen. Weitere Informationen finden Sie unter Verwalten der Datenqualität mit Pipelineerwartungen.

SDP erweitert die Funktionalität des Delta Lake. Da Tabellen, die von Pipelines erstellt und verwaltet werden, Delta-Tabellen sind, verfügen sie über die gleichen Garantien und Features, die von Delta Lake bereitgestellt werden. Siehe Was ist Delta Lake? in Azure Databricks.

Pipelines fügen neben den vielen Tabelleneigenschaften, die in Delta Lake festgelegt werden können, noch zusätzliche Tabelleneigenschaften hinzu. Siehe Referenz zu Pipelineeigenschaften und Referenz zu Delta-Tabelleneigenschaften.

Wie Tabellen durch Pipelines erstellt und verwaltet werden

Azure Databricks verwaltet automatisch Tabellen, die von Pipelines erstellt wurden, und bestimmt, wie Updates verarbeitet werden müssen, um den aktuellen Status einer Tabelle ordnungsgemäß zu berechnen und eine Reihe von Wartungs- und Optimierungsaufgaben auszuführen.

Für die meisten Vorgänge sollten Sie der Pipeline erlauben, alle Aktualisierungen, Einfügungen und Löschvorgänge in einer Zieltabelle zu verarbeiten. Ausführliche Informationen und Einschränkungen finden Sie unter Manuelle Löschungen oder Aktualisierungen beibehalten.

Wartungsaufgaben, die von Pipelines ausgeführt werden

Azure Databricks führt Wartungsaufgaben für Tabellen durch, die von Pipelines verwaltet werden, in einem optimalen Rhythmus mithilfe der Predictive Optimization. Die Wartung kann die Abfrageleistung verbessern und Kosten reduzieren, indem alte Versionen von Tabellen entfernt werden. Dazu gehören vollständige OPTIMIZE-Vorgänge und VACUUM-Vorgänge. Wartungsaufgaben werden in einem Zeitplan durchgeführt, der durch die Vorhersageoptimierung entschieden wird, und nur, wenn seit der vorherigen Wartung ein Pipelineupdate ausgeführt wurde.

Informationen dazu, wie oft eine predictive Optimierung ausgeführt wird, und um Wartungskosten zu verstehen, finden Sie in der Referenz zur Predictive Optimization System-Tabelle.

Einschränkungen

Eine Liste der Einschränkungen finden Sie unter Pipelinebeschränkungen.

Eine Liste der Anforderungen und Einschränkungen, die für die Verwendung von Pipelines mit Unity-Katalog spezifisch sind, finden Sie unter Verwenden des Unity-Katalogs mit Pipelines

Weitere Ressourcen

  • Last updated on