Databricks Streaming Tables mit Delta Live Tables bauen

Echtzeit-Daten-Pipelines sind das Rückgrat moderner Data-Plattformen geworden. Egal ob du User-Behavior trackst, IoT-Sensordaten ingestest oder geopolitische Events überwachst — die Fähigkeit, Daten zu verarbeiten, während sie ankommen (statt Stunden später), ist heute ein Wettbewerbsvorteil.

Delta Live Tables (DLT) ist Databricks' deklaratives Framework für zuverlässige, wartbare und testbare Data-Pipelines. In diesem Guide bauen wir produktionsreife Streaming Tables von Null auf — Ingestion, Transformation, Monitoring und Operational Best Practices.

TL;DR

• DLT abstrahiert Spark Structured Streaming weg — du deklarierst was, DLT regelt das wie.
• Drei Object-Types: Streaming Tables (append-only), Materialized Views (geplant/getriggert), Live Tables (Legacy-Begriff, jetzt unter Materialized Views).
• Medallion-Pattern: Bronze (Raw) → Silver (Validated) → Gold (Aggregated) — DLT macht das idiomatisch einfach.
• Kosten: Enhanced Autoscaling + triggered Mode wo möglich, sonst skaliert die DBU-Rechnung.

Was sind Delta Live Tables?

Delta Live Tables ist ein Framework, mit dem du deine Data-Pipeline in Python oder SQL definierst. Statt Spark-Structured-Streaming-Jobs manuell zu verwalten, deklarierst du, wie die Daten aussehen sollen — und DLT regelt Orchestrierung, Checkpointing, Schema-Evolution und Error-Handling.

DLT-Pipelines bestehen aus drei Hauptobjekttypen:

• Streaming Tables — append-only Tabellen, die aus Streaming-Sources ingesten (Kafka, Auto Loader etc.).
• Materialized Views — berechnete Tabellen, die geplant oder getriggert refreshen.
• Live Tables — Legacy-Begriff, in neueren DLT-Versionen unter Materialized Views vereint.

Deine erste DLT-Pipeline aufsetzen

Voraussetzungen

• Ein Databricks-Workspace (Unity Catalog empfohlen).
• Eine Streaming-Source — wir nutzen Auto Loader (Databricks' Cloud-File-Ingestion), der von Azure Data Lake Storage liest.

Schritt 1: Auto Loader als Source konfigurieren

Auto Loader nutzt Structured Streaming unter der Haube. Er ingestet inkrementell neue Files, wenn sie in einer Cloud-Storage-Location landen — ohne dass du selbst tracken musst, welche Files bereits verarbeitet wurden.

import dlt
from pyspark.sql.functions import col, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# Define the schema for incoming JSON events
event_schema = StructType([
    StructField("event_id", StringType(), True),
    StructField("event_type", StringType(), True),
    StructField("source", StringType(), True),
    StructField("location", StringType(), True),
    StructField("severity", DoubleType(), True),
    StructField("event_time", StringType(), True),
    StructField("raw_payload", StringType(), True),
])

@dlt.table(
    name="raw_events_bronze",
    comment="Bronze layer: raw events ingested via Auto Loader",
    table_properties={"quality": "bronze"}
)
def raw_events_bronze():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.schemaLocation", "/mnt/checkpoints/raw_events_schema")
            .option("cloudFiles.inferColumnTypes", "true")
            .load("/mnt/landing/events/")
            .withColumn("_ingested_at", current_timestamp())
            .withColumn("_source_file", col("_metadata.file_path"))
    )

Schritt 2: Silver-Layer — saubere und validierte Daten

Der Silver-Layer wendet Data-Quality-Expectations an. DLT's @dlt.expect-Decorators lassen dich Regeln deklarieren, die bei schlechten Records entweder warnen, droppen oder failen.

@dlt.table(
    name="events_silver",
    comment="Silver layer: validated and cleaned events",
    table_properties={"quality": "silver"}
)
@dlt.expect_or_drop("valid_event_id", "event_id IS NOT NULL")
@dlt.expect_or_drop("valid_severity", "severity BETWEEN 0.0 AND 10.0")
@dlt.expect("known_event_type", "event_type IN ('conflict', 'natural_disaster', 'economic', 'political')")
def events_silver():
    return (
        dlt.read_stream("raw_events_bronze")
            .select(
                col("event_id"),
                col("event_type"),
                col("source"),
                col("location"),
                col("severity"),
                col("event_time").cast("timestamp").alias("event_time"),
                col("raw_payload"),
                col("_ingested_at"),
            )
    )

Schritt 3: Gold-Layer — aggregierte Business-Metriken

Gold-Layer-Tabellen sind häufig Materialized Views (Batch-berechnet), aber du kannst auch Streaming-Gold-Tabellen für Real-Time-Aggregationen bauen.

@dlt.table(
    name="events_by_region_gold",
    comment="Gold layer: event counts and average severity by region (1-hour windows)",
    table_properties={"quality": "gold"}
)
def events_by_region_gold():
    from pyspark.sql.functions import window, avg, count

    return (
        dlt.read_stream("events_silver")
            .withWatermark("event_time", "10 minutes")
            .groupBy(
                window(col("event_time"), "1 hour"),
                col("location"),
                col("event_type")
            )
            .agg(
                count("event_id").alias("event_count"),
                avg("severity").alias("avg_severity")
            )
            .select(
                col("window.start").alias("window_start"),
                col("window.end").alias("window_end"),
                col("location"),
                col("event_type"),
                col("event_count"),
                col("avg_severity")
            )
    )

Pipeline über die CLI deployen

Du kannst DLT-Pipelines über die Databricks-CLI oder REST-API deployen. Hier eine pipeline.json-Konfiguration:

{
  "name": "harbinger-events-pipeline",
  "target": "harbinger_prod",
  "catalog": "main",
  "clusters": [
    {
      "label": "default",
      "autoscale": {
        "min_workers": 2,
        "max_workers": 8,
        "mode": "ENHANCED"
      }
    }
  ],
  "libraries": [
    {"notebook": {"path": "/Repos/harbinger/pipelines/events_pipeline"}}
  ],
  "continuous": true,
  "channel": "CURRENT",
  "edition": "ADVANCED"
}

Deploy via CLI:

databricks pipelines create --json @pipeline.json

Pipeline starten:

databricks pipelines start --pipeline-id <YOUR_PIPELINE_ID>

Streaming Tables monitoren

DLT exposed Metriken über das Event Log — eine Delta-Tabelle, die Pipeline-Runs, Data-Quality-Ergebnisse und Performance-Metriken aufzeichnet.

-- Query the DLT event log for data quality failures
SELECT
  timestamp,
  details:flow_progress.data_quality.dropped_records AS dropped_records,
  details:flow_progress.metrics.num_output_rows AS output_rows
FROM event_log('<pipeline_id>')
WHERE event_type = 'flow_progress'
  AND details:flow_progress.status = 'COMPLETED'
ORDER BY timestamp DESC
LIMIT 50;

Für Real-Time-Monitoring richtest du Alerts über Databricks-SQL-Alerts ein oder integrierst in deinen existierenden Observability-Stack (Datadog, Grafana etc.).

Fortgeschrittene Patterns

Schema-Evolution

DLT regelt Schema-Evolution automatisch, sobald du sie aktivierst:

@dlt.table(
    schema_hints="event_id STRING, event_type STRING",
    table_properties={"delta.enableChangeDataFeed": "true"}
)

Change Data Capture (CDC) mit APPLY CHANGES INTO

Für CDC-Szenarien (z. B. Sync aus einer transaktionalen Datenbank via Debezium):

dlt.apply_changes(
    target="events_scd1",
    source="raw_cdc_stream",
    keys=["event_id"],
    sequence_by=col("_commit_timestamp"),
    apply_as_deletes=col("_op") == "DELETE",
    except_column_list=["_op", "_commit_timestamp"]
)

Häufige Fallstricke

Cost-Optimization-Tipps

1. Enhanced Autoscaling nutzen — DLT's Enhanced Autoscaler ist bei Streaming-Workloads cleverer als Standard-Autoscaling.
2. Triggered vs Continuous — nur Continuous, wenn Sub-Minuten-Latency wirklich nötig ist.
3. Serverless DLT — jetzt GA; eliminiert Cluster-Management-Overhead und ist bei variablen Workloads oft günstiger.
4. Cluster richtig dimensionieren — beobachte num_output_rows gegen Cluster-Kosten; viele Streaming-Workloads laufen problemlos auf 2–4 Workern.

FAQ

Brauche ich Unity Catalog für DLT?

Nicht zwingend, aber stark empfohlen. Ohne UC bekommst du keine spaltenweise Lineage über DLT-Pipelines hinweg und kein einheitliches Access-Control. Für neue Production-Setups starte direkt mit UC.

Wie schnell ist DLT in der Praxis?

Im Triggered Mode ist die Latency dominiert vom Trigger-Intervall (typisch 1–10 Minuten). Im Continuous Mode liegen End-to-End-Latenzen bei 10–60 Sekunden, abhängig von Cluster-Größe und Daten-Volumen. Sub-Sekunden-Latency ist nicht das Ziel von DLT — dafür Apache Flink oder Materialize.

Kann ich DLT-Pipelines mit Airflow orchestrieren?

Ja, über die Databricks-CLI oder REST-API. Eine gängige Pattern: Airflow triggert DLT-Pipelines stündlich, DLT übernimmt die eigentliche Datenverarbeitung. Alternativ Databricks Workflows (Native Orchestrator), das tiefer in DLT integriert ist als Airflow.

Was passiert mit DLT-Tabellen, wenn ich die Pipeline lösche?

Managed Tables werden mit der Pipeline gelöscht — Daten weg. External Tables (auf eigenem Cloud-Storage) bleiben erhalten. Für Production immer External Tables mit klaren Storage-Pfaden, damit Pipeline-Recreation keine Daten zerstört.

Fazit

Delta Live Tables abstrahiert die Komplexität von Spark-Structured-Streaming-Checkpoints, Schema-Evolution und Data Quality weg — und lässt dein Team sich auf Business-Logik konzentrieren statt auf Infrastruktur-Plumbing. Ob Real-Time-Fraud-Detection oder Event-Stream-Ingestion: DLT gibt dir ein zuverlässiges Fundament.

Stand: Mai 2026. Databricks ändert Preise und Features regelmäßig — verifiziere kritische Annahmen direkt bei Databricks.