Komponententests für Pipelines

Important

Dieses Feature befindet sich in der Betaversion.

Allgemeine Informationen zu Python Komponententests in Databricks finden Sie unter Python Komponententests.

Lakeflow-Pipelines ermöglichen das Schreiben von Python-Komponententests im webbasierten Lakeflow Pipelines Editor. Auf diese Weise können Sie Python- oder SQL-Transformationslogik mithilfe von simulierten Daten überprüfen. Mit dem Framework zum Testen von Pipelines können Sie Randfälle testen, proprietäre Pipeline-APIs (Auto CDC, Streamingtabellen, Erwartungen, Append-Flows) validieren und mithilfe von Mock-Eingaben für unterstützte Operationen mit Tabellenbezeichnern iterieren. Überprüfen Sie die Isolationseinschränkungen vor dem Ausführen von Tests.

• Isolierte Testausführung: Das Framework stellt eine SparkSession bereit, die Tabellenvorgänge in ein temporäres Testschema im Standardkatalog der Pipeline umleitet, sodass Sie Eingabedaten simuliert und Testausgaben schreiben können, ohne dass sich dies auf Produktionstabellen auswirkt. Isolation gilt für Vorgänge, die nach Namen auf eine Tabelle verweisen; siehe Einschränkungen.
• Flexibler Testbereich: Führen Sie mithilfe der Test-SparkSession eine Teilmenge einer Pipeline (einzelne Tabellen, Ketten abhängiger Tabellen oder die gesamte Pipeline) mit den Rechenressourcen der Pipeline aus.
• Ergebnisüberprüfung: Überprüfen Sie die Ergebnisse isolierter Ausgabetabellen, die in einem Test mithilfe von Standard-Pytest-Assertionen erstellt wurden.

Wann Unit-Tests eingesetzt werden sollten

Typische Anwendungsfälle sind:

• Validierung neuer Transformationslogik: Prüfen Sie, ob Ihre Transformation das erwartete Schema, die Zeilenanzahl, Aggregationen und die Geschäftslogik liefert, bevor Sie sie auf Produktionsdaten anwenden.
• Testen von Auto CDC-Spezifikationen: Überprüfen Sie, ob Ihre Auto CDC-Flussdefinitionen Änderungsereignisse ordnungsgemäß verarbeiten, Einfügungen, Aktualisierungen, Löschungen und SCD-Typen (Langsam ändernde Dimension) mithilfe von Pseudodaten verarbeiten.
• Erwartungen und Datenqualitätsregeln testen: Überprüfen Sie, ob Erwartungen dann fehlschlagen, wenn sie fehlschlagen sollten, und erfolgreich sind, wenn die Daten gültig sind.
• Tests über abhängige Tabellen hinweg: Testen Sie Transformationsketten (z. B. Bronze, Silber und Gold), um zu überprüfen, ob die Daten korrekt durch Ihr Pipeline-Diagramm fließen.

Requirements

• Owner sowie die für den Standardkatalog der Pipeline. Das Framework benötigt diese Berechtigungen, um das temporäre Testschema zu erstellen, in dem Tests ausgeführt werden.

  Um die Berechtigung für die Pipeline zu überprüfen oder festzulegen, öffnen Sie die Pipeline und klicken Sie auf Freigeben. Sie müssen die Pipeline Owner (IS OWNER) verwenden; CAN RUN und CAN MANAGE reichen nicht aus, um Tests auszuführen. Siehe Konfigurieren von Pipelineberechtigungen.

  Um die Katalogberechtigungen zu überprüfen oder festzulegen, öffnen Sie den Katalog im Katalog-Explorer, wählen Sie die Registerkarte "Berechtigungen" aus, und bestätigen Sie, dass Sie verfügen USE CATALOG und .CREATE SCHEMA Ein Katalogbesitzer, ein Metastore-Administrator oder ein Benutzer mit der Berechtigung MANAGE kann diese auch per SQL vergeben:

  GRANT USE CATALOG, CREATE SCHEMA ON CATALOG <catalog_name> TO `<principal>`;
  

  Weitere Informationen finden Sie unter Unity Catalog-Berechtigungsreferenz.

• Die Pipeline muss im triggergesteuerten Modus (nicht kontinuierlich) konfiguriert werden.

• Die Pipeline muss sich auf dem PREVIEW-Kanal befinden. Unit-Tests sind in der Beta und nur auf PREVIEW verfügbar.

• Spark Connect wird nicht unterstützt.

Hinweis

Die Testisolation umfasst Tabellenvorgänge, die nach Namen auf eine Tabelle verweisen. Vorgänge, die die Isolation umgehen, können sowohl im Testcode als auch in jedem pipelinecode auftreten, der von den ausgewählten Ausgaben ausgeführt wird, einschließlich der transitiven Abhängigkeiten. Eine Testdatei, die sicher aussieht, kann dennoch einen Pipeline-Ablauf ausführen, der über einen Pfad oder Connector liest oder schreibt und dabei auf Produktionsdaten zugreift. Befolgen Sie die folgenden Regeln, um Tests daran zu hindern, dass sich Dies auf Produktionsdaten oder Metadaten auswirkt:

• Verweisen Sie auf jede Tabelle anhand des Namens (catalog.schema.table), und modellieren Sie alle Eingaben anhand des Namens. Lesen oder schreiben Sie nicht über einen Pfad (/Volumes/..., dbfs:/..., s3://..., abfss://...) und lesen Sie nicht aus Konnektoren wie Kafka oder Auto Loader. Diese umgehen die Isolation und wirken auf reale Produktionssysteme ein.
• Führen Sie keine Governance- oder Eigentümerangaben aus, wie GRANT, REVOKE, ALTER ... OWNER TO, SET/UNSET TAGS oder CREATE/DROP POLICY. Diese werden für das echte Produktionssicherungsobjekt ausgeführt.
• Erstellen Sie keine Kataloge oder Schemas (CREATE CATALOG, CREATE SCHEMA). Diese erreichen Ihren tatsächlichen Unity Catalog-Metastore.
• Führen Sie nicht die gesamte Pipeline aus, wenn ihr Graph pfadbasierte Eingaben, Konnektoren, imperative Schreiboperationen oder andere externe Nebenwirkungen enthält. Wählen Sie nur Ausgaben aus, deren Abhängigkeiten unterstützte Katalogtabellenvorgänge verwenden und durch simulierte Eingaben ersetzt wurden.

Ausführliche Informationen finden Sie unter Einschränkungen.

Einschränkungen

Warning

Einige Vorgänge umgehen die Testisolation und können auf echte Produktionsdaten oder Metadaten reagieren. Überprüfen Sie die folgenden Einschränkungen, bevor Sie Tests ausführen.

Testisolation erfolgt nur anhand des Tabellennamens

• Lesen oder schreiben Sie nicht über Pfad oder Konnektor. Isolation leitet nur Vorgänge um, die über ihren Namen auf eine Tabelle verweisen (zum Beispiel spark.read.table("catalog.schema.table") oder df.write.saveAsTable("catalog.schema.table")). Vorgänge, die über einen Pfad oder einen Connector angesprochen werden, umgehen die Isolation und wirken direkt auf reale Produktionssysteme:

  • Schreiben über einen Pfad (z. B. einen df.write.save("/Volumes/...")dbfs:/-Pfad oder einen Cloud- oder External-Location-Pfad wie s3://... oder abfss://...) schreibt in den realen Produktionsspeicher und kann Produktionsdaten überschreiben.
  • Das Lesen über den Pfad (z. B. spark.read.load(path) oder spark.read.format("delta").load(path)) gibt echte Produktionsdaten statt Ihres Mocks zurück.
  • Das Lesen von einem Connector stellt eine Verbindung mit der realen Produktionsquelle her. Dazu gehören Kafka (liest aus den echten Brokern) und Auto Loader (cloudFiles, der aus dem echten Cloudspeicherpfad liest). Keines wird zu Ihren simulierten Daten umgeleitet.

• Verwenden Sie die tabellenwertige Funktion event_log() nicht in einem Pipeline-Unit-Test. Im Testmodus wird event_log() nicht in das Ereignisprotokoll Ihres Testlaufs umgeleitet. Sie kann das Produktions- oder zuvor registrierte Ereignisprotokoll zurückgeben, sodass Assertionen damit Produktionsdaten lesen können. Verwenden Sie stattdessen die von der Ausführung zurückgegebene event_log_table_name und fragen Sie sie über test_spark ab. event_log_table_name kann None sein (z. B. wenn der Name der Ereignisprotokolltabelle nicht aufgelöst werden kann). Prüfen Sie dies also vor der Abfrage:

  status = test_pipeline.run(test_spark, set(["catalog.schema.table"]))
  assert status.event_log_table_name is not None
  events = test_spark.table(status.event_log_table_name)
  

  Bestätigen Sie nicht status.is_success , bevor Sie das Ereignisprotokoll lesen, wenn Ihr Ziel darin besteht, ein fehlgeschlagenes Update zu diagnostizieren. Das Ereignisprotokoll wird häufig untersucht, um zu verstehen, warum ein Update fehlgeschlagen ist.

Governance und DDL-Operationen

• Katalog-, Schema-, Berechtigungs-, Besitzer-, Tag- und Richtlinienmutationen werden nicht unterstützt. Dazu gehören CREATE/DROP/ALTER CATALOG, (CREATE/DROP/ALTER SCHEMAeinschließlich SET MANAGED LOCATION), GRANT/REVOKE, , ALTER ... OWNER TO,SET/UNSET TAGS und .CREATE/DROP POLICY Einige Formen von SQL-Anweisungen, die über test_spark ausgeführt werden, werden im Sinne einer mehrschichtigen Sicherheitsstrategie zurückgewiesen; andere Formen oder dieselben Vorgänge, die über direkte APIs aufgerufen werden, können auf echte Produktionsobjekte zugreifen. Verlassen Sie sich nicht auf diese Wächter als Isolationsgrenze. Verwenden Sie diese Anweisungen weder in Ihrem Testcode noch in Pipeline-Code, der von den ausgewählten Ausgaben ausgeführt wird.

Betriebsbeschränkungen

• Die gleichzeitige Ausführung wird nicht unterstützt: Das Gleichzeitige Ausführen eines Tests und eines Pipelineupdates wird nicht unterstützt, und das System verhindert es nicht. Es gibt keine Koordination zwischen den beiden, sodass sie gleichzeitig für Ressourcen kämpfen können, die Leistung Ihres Produktionsupdates stark beeinträchtigen oder der Test nicht gestartet werden kann. Starten Sie keinen Test, während die Pipeline ein Update ausführt (oder ein Update startet, während ein Test ausgeführt wird); Warten Sie, bis alle laufenden Updates abgeschlossen sind, bevor Sie Tests ausführen.
• Temporäre Schemas nach einer ungewöhnlichen Beendigung: Jeder Testlauf erstellt ein temporäres Schema (mit dem Namen redirecting_<id>) im Standardkatalog der Pipeline und löscht es automatisch, wenn der Lauf abgeschlossen ist. Wenn eine Ausführung abnormal endet (zum Beispiel, wenn die Recheninstanz während der Ausführung ausfällt), kann das temporäre Schema zurückbleiben und die Mock- und Ausgabetabellen der Ausführung enthalten. Dies wirkt sich nicht auf Produktionsdaten aus. Um Speicherplatz freizugeben, löschen Sie im Standardkatalog der Pipeline manuell alle verbleibenden Schemas, deren Namen mit redirecting_ beginnen.
• Testläufe verbrauchen Rechenkapazität: Testläufe werden auf der Rechenkapazität der Pipeline ausgeführt und als normale Pipeline-Updates in Rechnung gestellt. Es gibt keine separate Meterung für Testläufe.
• Die vollständige Aktualisierung wird nicht unterstützt: Nur selektive Aktualisierung ist verfügbar. test_pipeline.run() aktualisiert die von Ihnen ausgewählten Outputs (oder alle Outputs, wenn keine Auswahl übergeben wird); Full Refresh und die Auswahl für Full Refresh sind nicht implementiert.

Einschränkungen bei Erstellung und Originaltreue

• Nur Editor-Ausführung: Tests müssen über den webbasierten Lakeflow-Pipelines-Editor ausgeführt werden.
• nur Python Tests: Tests müssen in Python geschrieben werden. Sie können SQL-Pipelines testen, aber die Tests selbst müssen in Python geschrieben werden.
• Governance-Konformität: Mock-Daten übernehmen keine Zeilenfilter oder Spaltenmasken, die für die Produktionstabellen definiert sind, die sie ersetzen. Testergebnisse spiegeln die simulierten Eingaben genau so wider, wie sie bereitgestellt werden, und können sich davon unterscheiden, wie sich dieselbe Abfrage bei gesteuerten Produktionsdaten verhält.

Schritt 1: Aktualisieren von Pipelineeinstellungen

Konfigurieren Sie die Pipeline so, dass sie im PREVIEW-Kanal im Trigger-Modus ausgeführt wird.

1. Öffnen Sie in der Benutzeroberfläche Ihre Pipeline, und klicken Sie auf Einstellungen>Erweiterte Einstellungen>Kanal>Vorschau
2. Legen Sie den Pipelinemodus auf "Ausgelöst" fest (verwenden Sie "Fortlaufend" nicht).

Alternativ können Sie die JSON-Pipelineeinstellungen direkt bearbeiten:

"continuous": false,
"channel": "PREVIEW"

Schritt 2: Erstellen einer Testdatei

Klicken Sie im Lakeflow Pipelines-Editor auf die + Schaltfläche (hinzufügen), und wählen Sie "Testen" aus. Dadurch wird eine Testdatei erstellt (sowie der Ordner tests, falls er noch nicht vorhanden ist), die nicht im Quellcode Ihrer Pipeline enthalten ist. Sie müssen den tests Ordner nicht selbst erstellen.

Schritt 3: Generieren von Tests

Genie Code kann Testgerüste generieren:

• Klicken Sie in der Testdatei auf die Schaltfläche " Tests generieren ".

  

• Alternativ verwenden Sie /tests im Genie Code-Agent-Modus.

  

Verwenden Sie Genie Code, um Boilerplate-Code zu generieren, und passen Sie ihn dann für Ihre Randfälle an.

Alternativ können Sie den Testcode selbst schreiben. Fügen Sie die folgenden Importe am Anfang jeder Testdatei hinzu:

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

Schritt 4: Ausführen von Tests

Führen Sie Tests aus dem Lakeflow Pipelines Editor aus:

• Klicken Sie auf das (Play) im Randbereich neben einer Testfunktion, um einen einzelnen Test auszuführen.
• Klicken Sie oben in der Testdatei auf "Tests ausführen ", um alle Tests in dieser Datei auszuführen.

Testergebnisse (Erfolg oder Fehler) werden im unteren Bereich des Editors angezeigt. Überprüfen Sie Assertionsfehler zum Debuggen von Fehlern.

Testen von APIs

API	Description
TestPipeline.active()	Gibt ein TestPipeline Objekt für die Pipeline zurück, die derzeit im Lakeflow Pipelines Editor bearbeitet wird. Dieses Objekt ist ein Verweis auf die Pipeline, einschließlich Quellcode, Konfigurationen, Standardkatalog/Schema usw.
test_pipeline.run(test_spark, set([table_names]))	Führt synchron eine Aktualisierung der Pipeline aus und führt eine selektive Aktualisierung durch, wenn Tabellennamen angegeben sind. Wird nach der Pipelineausführung zurückgegeben, wenn diese erfolgreich abgeschlossen wurde oder mit einer Ausnahme endet.
test_spark Vorrichtung	Erstellt eine Test-SparkSession mit Katalog-Tabellenumleitung, die Lese- und Schreibvorgänge für Tabellen, die auf eine Tabelle über den Namen verweisen (zum Beispiel spark.read.table("catalog.schema.table") oder df.write.saveAsTable("catalog.schema.table")), automatisch an ein temporäres Testschema umleitet. Umleitung gilt nur für namenbasierte Tabellenvorgänge; Es deckt keine Lese- oder Schreibvorgänge ab, die vom Pfad oder über einen Verbinder adressiert werden, die direkt auf das eigentliche System reagieren. Informationen finden Sie unter Einschränkungen.

Erstellen von Simulierten Daten

Sie können Eingabedaten mithilfe von SQL oder createDataFrame:

# Option 1: Using SQL
test_spark.sql("""
    CREATE TABLE catalog.schema.table_name AS
    SELECT * FROM VALUES
        (1, 'value1'),
        (2, 'value2')
    AS t(id, name)
""")

# Option 2: Using createDataFrame
df = test_spark.createDataFrame(
    [(1, 'value1'), (2, 'value2')],
    schema=["id", "name"]
)
df.write.saveAsTable("catalog.schema.table_name")

Um größere Mengen realistischer synthetischer Daten zu generieren, können Sie die Faker-Bibliothek verwenden. Führen Sie zuerst %pip install faker in Ihrer Pipeline aus und erstellen Sie dann einen DataFrame aus Faker-gestützten UDFs:

# Option 3: Using Faker for synthetic data
from pyspark.sql import functions as F
from faker import Faker

fake = Faker()
fake_firstname = F.udf(fake.first_name)
fake_lastname = F.udf(fake.last_name)
fake_email = F.udf(fake.ascii_company_email)

df = (
    test_spark.range(0, 100)
    .withColumn("firstname", fake_firstname())
    .withColumn("lastname", fake_lastname())
    .withColumn("email", fake_email())
)
df.write.saveAsTable("catalog.schema.table_name")

Führen Sie die Pipeline oder bestimmte Tabellen aus

# Run specific tables
test_pipeline.run(test_spark, set(["catalog.schema.table1", "catalog.schema.table2"]))

# Run all tables in the pipeline
test_pipeline.run(test_spark)

Beispiele

Beispiel 1: Testen von Aggregationen mit Zeilenanzahl, Schema und Nullbehandlung

Ziel: Überprüfen, ob die Benutzeraggregation Benutzer nach Typ korrekt zählt, Null-E-Mail-Adressen korrekt verarbeitet und das erwartete Schema erzeugt.

Pipelinetransformationen:

Diese Transformationen erstellen eine einfache Zwei-Tabellen-Pipeline: users wählt Benutzerdaten aus und counts gruppiert Benutzer nach Typ und Zählt die Gesamtzahl der Benutzer und gültige E-Mails.

from pyspark import pipelines as dp
from pyspark.sql.functions import col, count, count_if

@dp.table
def users():
    return (
        spark.read.table("catalog.schema.wanderbricks_users")
        .select("user_id", "email", "name", "user_type")
    )

@dp.table
def counts():
    return (
        spark.read.table("catalog.schema.users")
        .withColumn("valid_email", col("email").isNotNull())
        .groupBy("user_type")
        .agg(
            count("user_id").alias("total_count"),
            count_if("valid_email").alias("count_valid_emails")
        )
    )

Tests:

Diese Tests überprüfen Zeilenanzahl, Schemastruktur, Nullbehandlung und Aggregationslogik, indem simulierte Benutzerdaten mit absichtlichen Nullen erstellt und die Pipeline isoliert ausgeführt wird.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark
from pyspark.testing import assertDataFrameEqual

test_pipeline = TestPipeline.active()

# Mock data fixture
def mock_users(session):
    session.sql("""
        CREATE TABLE catalog.schema.wanderbricks_users AS
        SELECT * FROM VALUES
            (1, 'alice@example.com', 'Alice', 'admin'),
            (2, NULL, 'Bob', 'user'),
            (3, 'charlie@example.com', 'Charlie', 'user'),
            (4, NULL, 'Dana', 'admin')
        AS t(user_id, email, name, user_type)
    """)

# Test 1: Row count
def test_users_row_count(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    assert result.count() == 4

# Test 2: Schema validation
def test_users_schema(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    expected_fields = {"user_id", "email", "name", "user_type"}
    actual_fields = set(f.name for f in result.schema.fields)
    assert expected_fields == actual_fields

# Test 3: Null handling
def test_users_null_handling(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    null_emails = result.filter("email IS NULL").count()
    assert null_emails == 2

# Test 4: Aggregation
def test_counts(test_spark):
    mock_users(test_spark)
    # Run both tables since counts depends on users
    test_pipeline.run(test_spark, set(["catalog.schema.users", "catalog.schema.counts"]))
    result = test_spark.table("catalog.schema.counts")
    # Check counts for each user_type
    admin_row = result.filter("user_type = 'admin'").collect()[0]
    user_row = result.filter("user_type = 'user'").collect()[0]
    assert admin_row["total_count"] == 2
    assert admin_row["count_valid_emails"] == 1
    assert user_row["total_count"] == 2
    assert user_row["count_valid_emails"] == 1

# Test 5: Full DataFrame comparison with assertDataFrameEqual
def test_counts_full_dataframe(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users", "catalog.schema.counts"]))
    result = test_spark.table("catalog.schema.counts")
    expected = test_spark.createDataFrame(
        [("admin", 2, 1), ("user", 2, 1)],
        schema=["user_type", "total_count", "count_valid_emails"]
    )
    assertDataFrameEqual(result, expected)

Beispiel 2: Testen des automatischen CDC

Ziel: Überprüfen Sie, ob Auto CDC den Änderungsfeed mit Einfügungen und Aktualisierungen ordnungsgemäß verarbeitet.

Pipelinetransformation:

Diese Transformation richtet Auto CDC anhand eines Änderungsfeeds ein, der fortlaufende Änderungen liest und sie als SCD-Typ 1 auf die Zieltabelle anwendet, wobei nur die neueste Version beibehalten wird.

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.view
def users():
    return spark.readStream.table("catalog.schema.change_feed")

dp.create_streaming_table("target_autocdc")
dp.create_auto_cdc_flow(
    target="target_autocdc",
    source="users",
    keys=["userId"],
    sequence_by=col("ts"),
    stored_as_scd_type=1
)

Tests:

Der erste Test erstellt einen Simulierten Änderungsfeed mit mehreren Datensätzen für dasselbe userId (Simulieren eines Updates) und überprüft, ob nur der neueste Datensatz im Ziel aufbewahrt wird. Der zweite Test simuliert spät eintreffende und nicht in der richtigen Reihenfolge eintreffende Ereignisse, indem er die Pipeline ausführt, dem Änderungsfeed weitere Ereignisse hinzufügt und die Pipeline erneut ausführt.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

# Test 1: Standard inserts and updates
def test_auto_cdc_flow(test_spark):
    # Create a mock change feed table
    test_spark.sql("""
        CREATE TABLE catalog.schema.change_feed AS
        SELECT * FROM VALUES
            (1, 'Alice', 1000),
            (2, 'Bob', 1001),
            (1, 'Alice Updated', 1002)
        AS t(userId, name, ts)
    """)
    # Run the pipeline
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))
    # Read the output
    result = test_spark.table("catalog.schema.target_autocdc")
    # Verify two users exist
    user_ids = set(row["userId"] for row in result.collect())
    assert user_ids == {1, 2}
    # Verify latest record for userId=1 has ts=1002
    latest_user1 = result.filter("userId = 1").collect()[0]
    assert latest_user1["ts"] == 1002
    assert latest_user1["name"] == "Alice Updated"
    # Verify userId=2 has ts=1001
    user2 = result.filter("userId = 2").collect()[0]
    assert user2["ts"] == 1001

# Test 2: Late-arriving and out-of-order events
def test_auto_cdc_late_arriving(test_spark):
    # First batch of change events
    test_spark.sql("""
        CREATE TABLE catalog.schema.change_feed AS
        SELECT * FROM VALUES
            (1, 'Alice', 1000),
            (2, 'Bob', 1001)
        AS t(userId, name, ts)
    """)
    # Run the pipeline with the initial batch
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))

    # Append late-arriving events to the change feed:
    # - A newer event for userId=1 (ts=1003) that arrived after the first run
    # - A stale event for userId=2 (ts=999) with a timestamp older than what is already applied
    test_spark.sql("""
        INSERT INTO catalog.schema.change_feed VALUES
            (1, 'Alice Updated', 1003),
            (2, 'Bob (stale)', 999)
    """)
    # Re-run the pipeline. sequence_by=ts ensures stale events do not overwrite newer state.
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))

    result = test_spark.table("catalog.schema.target_autocdc")
    # userId=1 should reflect the newer late-arriving event
    alice = result.filter("userId = 1").collect()[0]
    assert alice["ts"] == 1003
    assert alice["name"] == "Alice Updated"
    # userId=2 should be unchanged: the stale event with an older ts is ignored
    bob = result.filter("userId = 2").collect()[0]
    assert bob["ts"] == 1001
    assert bob["name"] == "Bob"

Beispiel 3: Test des automatischen CDC anhand einer Momentaufnahme

Ziel: Überprüfen Sie, ob CDC Momentaufnahmenänderungen ordnungsgemäß verarbeitet, einschließlich Einfügungen, Aktualisierungen und Löschvorgängen.

Pipelinetransformation:

Diese Transformation richtet Auto CDC aus einem Snapshot ein, das aus einer Snapshot-Tabelle liest und Änderungen im Zeitverlauf als SCD-Typ 2 verfolgt (der vollständige Verlauf bleibt erhalten).

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
    return spark.read.table("catalog.schema.snapshot")

dp.create_streaming_table("catalog.schema.target")
dp.create_auto_cdc_from_snapshot_flow(
    target="target",
    source="source",
    keys=["userId"],
    stored_as_scd_type=2
)

Test:

Dieser Test erstellt eine anfängliche Momentaufnahme, führt die Pipeline aus und simuliert dann eine Momentaufnahmeaktualisierung, indem neue Daten abgeschnitten und eingefügt werden, um sicherzustellen, dass CDC alle Änderungen erfasst.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

def test_auto_cdc_from_snapshot_flow(test_spark):
    # Create initial snapshot
    test_spark.sql("""
        CREATE TABLE catalog.schema.snapshot AS
        SELECT * FROM VALUES
            (1, 'Alice', '2024-01-01'),
            (2, 'Bob', '2024-01-02')
        AS t(userId, name, created_at)
    """)
    # Run the pipeline
    test_pipeline.run(test_spark, set(["catalog.schema.target"]))
    # Simulate a new snapshot by truncating and inserting updated data
    test_spark.sql("TRUNCATE TABLE catalog.schema.snapshot")
    test_spark.sql("INSERT INTO catalog.schema.snapshot VALUES (2, 'Bob', '2024-01-03')")
    test_pipeline.run(test_spark, set(["catalog.schema.target"]))
    # Verify SCD Type 2: should have 3 rows (original Alice, original Bob, updated Bob)
    result = test_spark.table("catalog.schema.target")
    assert result.count() == 3
    user_ids = [row["userId"] for row in result.collect()]
    assert set(user_ids) == {1, 2}

Beispiel 4: Testen von Verknüpfungen und Erwartungen

Ziel: Validieren, dass Joins korrekt funktionieren und dass Erwartungen ungültige Daten herausfiltern.

Pipelinetransformation:

Diese Transformation verknüpft Eigenschaftsbilder mit Annehmlichkeiten und wendet eine Erwartung an, Bilder herauszufiltern, die vor Januar 2024 hochgeladen wurden.

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("uploaded after Jan 2024", "uploaded_at > '2024-01-01'")
def property_images_amenities_join():
    return (
        spark.read.table("catalog.schema.property_images")
        .join(
            spark.read.table("catalog.schema.property_amenities"),
            on="property_id",
            how="inner"
        )
    )

Tests:

Diese Tests stellen sicher, dass die Verknüpfung die richtige Anzahl von Zeilen erzeugt und dass die Erwartung Datensätze erfolgreich mit ungültigen Uploaddaten herausfiltert.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

# Mock property datasets
def mock_properties(session):
    session.sql("""
        CREATE TABLE catalog.schema.property_images AS
        SELECT * FROM VALUES
            (101, 'img1.jpg', '2024-02-01'),
            (102, 'img2.jpg', '2024-01-15'),
            (103, 'img3.jpg', '2024-12-20')
        AS t(property_id, image_url, uploaded_at)
    """)
    session.sql("""
        CREATE TABLE catalog.schema.property_amenities AS
        SELECT * FROM VALUES
            (101, 'wifi'),
            (102, 'pool'),
            (103, 'parking')
        AS t(property_id, amenity)
    """)

# Test 1: Join
def test_property_join(test_spark):
    mock_properties(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.property_images_amenities_join"]))
    result = test_spark.table("catalog.schema.property_images_amenities_join")
    # Should have 3 rows after join
    assert result.count() == 3
    # Check all property_ids are present
    property_ids = set(row["property_id"] for row in result.collect())
    assert property_ids == {101, 102, 103}

# Test 2: Expectation
def test_property_expectation(test_spark):
    mock_properties(test_spark)
    # Add a row with uploaded_at before Jan 2024
    test_spark.sql("""
        INSERT INTO catalog.schema.property_images VALUES (104, 'img4.jpg', '2023-12-31')
    """)
    # Add a matching row in the amenities table for the join
    test_spark.sql("""
        INSERT INTO catalog.schema.property_amenities VALUES (104, 'gym')
    """)
    test_pipeline.run(test_spark, set(["catalog.schema.property_images_amenities_join"]))
    result = test_spark.table("catalog.schema.property_images_amenities_join")
    # Only property_ids with uploaded_at > '2024-01-01' should be present
    valid_ids = set(row["property_id"] for row in result.collect())
    assert 104 not in valid_ids
    assert valid_ids == {101, 102, 103}