Data Quality Testing: Praktischer Guide für Data Engineers

TL;DR: Schlechte Daten kosten mehr als schlechter Code. Eine kaputte Pipeline wirft einen Fehler — schlechte Daten korrumpieren Dashboards leise, oft wochenlang. Datenqualitäts-Tests an drei Layern (Ingestion-Schema, SQL-Assertions nach Transformation, statistische Checks bei Aggregaten) plus ein Quality-Gate, das die Pipeline blockt, ist das Minimum für Produktion.

Schlechte Daten kosten mehr als schlechter Code. Eine kaputte Pipeline wirft einen Fehler, und jemand fixt es. Schlechte Daten korrumpieren Dashboards leise, führen Stakeholder in die Irre und untergraben das Vertrauen in deine gesamte Data-Plattform — manchmal wochenlang, bevor jemand etwas merkt.

Data Quality Testing ist die Praxis, systematisch zu validieren, dass deine Daten an jeder Pipeline-Stufe die definierten Erwartungen erfüllen. Wer ohne das produktive Datenworkloads fährt, fliegt blind.

Warum Data Quality Testing zählt

Ein Szenario, das jeder Data Engineer kennt: eine Source-API ändert über Nacht ihr Response-Schema. Kein Fehler. Deine Pipeline ingestiert die Daten, transformiert sie, lädt sie ins Warehouse. Drei Tage später fragt eine VP, warum die Umsatzzahlen um 40 % gefallen sind. Antwort? Ein Feld wurde von total_amount zu amount_total umbenannt, und deine Transformation hat lautlos Nulls produziert.

Data Quality Testing fängt das beim Ingest ab — nicht im Board-Meeting.

Was kostet es, Quality-Tests wegzulassen?

Sechs Kerndimensionen der Datenqualität

Bevor du Tests schreibst, brauchst du ein Framework dafür, was getestet wird. Diese sechs Dimensionen decken die meisten realen Szenarien:

Eine Data-Quality-Testing-Strategie bauen

Layer 1: Schema-Validierung (Ingestion)

Strukturelle Probleme abfangen, bevor die Daten in die Pipeline gelangen. Das ist deine billigste Verteidigungslinie.

# Python — Schema validation with Pydantic
from pydantic import BaseModel, field_validator
from typing import Optional
from datetime import datetime

class OrderRecord(BaseModel):
    order_id: str
    customer_id: str
    amount: float
    currency: str
    created_at: datetime
    status: str

    @field_validator("amount")
    @classmethod
    def amount_must_be_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError(f"amount must be positive, got {v}")
        return v

    @field_validator("currency")
    @classmethod
    def currency_must_be_valid(cls, v: str) -> str:
        valid = {"EUR", "USD", "GBP", "CHF"}
        if v not in valid:
            raise ValueError(f"unexpected currency: {v}")
        return v

# Validate each record at ingestion
def validate_batch(records: list[dict]) -> tuple[list, list]:
    valid, invalid = [], []
    for record in records:
        try:
            validated = OrderRecord(**record)
            valid.append(validated.model_dump())
        except Exception as e:
            invalid.append({"record": record, "error": str(e)})
    return valid, invalid

Das fängt Type-Mismatches, fehlende Felder und Werte außer Range ab, bevor sie das Warehouse erreichen.

Layer 2: SQL-Assertions (Transformation)

Nachdem die Daten gelandet sind, lass Assertions gegen die transformierten Tabellen laufen. Das sind deine zentralen Quality-Gates.

-- PostgreSQL — Common data quality assertions

-- 1. Uniqueness: No duplicate order IDs
SELECT order_id, COUNT(*) AS cnt
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;
-- Expected: 0 rows

-- 2. Completeness: No null customer references
SELECT COUNT(*) AS null_customers
FROM orders
WHERE customer_id IS NULL;
-- Expected: 0

-- 3. Referential integrity: Every order references a valid customer
SELECT o.order_id, o.customer_id
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.id
WHERE c.id IS NULL;
-- Expected: 0 rows

-- 4. Freshness: Data updated within last 2 hours
SELECT
    CASE
        WHEN MAX(updated_at) < NOW() - INTERVAL '2 hours'
        THEN 'STALE'
        ELSE 'FRESH'
    END AS freshness_status
FROM orders;
-- Expected: FRESH

-- 5. Volume: Row count within expected range
SELECT
    CASE
        WHEN COUNT(*) BETWEEN 1000 AND 100000
        THEN 'OK'
        ELSE 'ANOMALY'
    END AS volume_check
FROM daily_orders
WHERE order_date = CURRENT_DATE;

Layer 3: Statistische Checks (Aggregation)

Für Metrik-Tabellen und Aggregationen geh über Row-Level-Checks hinaus. Schau nach Verteilungs-Anomalien.

# Python — Statistical quality checks with pandas
import pandas as pd
import numpy as np

def check_metric_quality(
    current: pd.DataFrame,
    historical: pd.DataFrame,
    metric_col: str,
    z_threshold: float = 3.0
) -> dict:
    """Flag metrics that deviate significantly from historical norms."""

    hist_mean = historical[metric_col].mean()
    hist_std = historical[metric_col].std()
    current_val = current[metric_col].sum()

    z_score = (current_val - hist_mean) / hist_std if hist_std > 0 else 0

    return {
        "metric": metric_col,
        "current_value": current_val,
        "historical_mean": round(hist_mean, 2),
        "z_score": round(z_score, 2),
        "status": "ANOMALY" if abs(z_score) > z_threshold else "OK",
    }

# Usage
# results = check_metric_quality(today_df, last_30_days_df, "revenue")
# if results["status"] == "ANOMALY":
#     alert_on_call_engineer(results)

Tooling: Was du nutzen solltest

Mehrere Frameworks machen Data Quality Testing leichter. Ehrlicher Vergleich:

Meine Einschätzung: Wenn du dbt nutzt, starte mit den eingebauten Tests und ergänze Great Expectations für alles, was dbt nicht ausdrücken kann. Ohne dbt hat Sodas YAML-Ansatz die sanfteste Lernkurve.

Häufige Fehler und Pitfalls

1. Nur in Produktion testen. Wenn deine Quality-Tests nach dem Warehouse-Load laufen, hast du schon schlechte Daten an Downstream-Konsumenten ausgeliefert. Teste an Ingestion-Grenzen.

2. Alert Fatigue. Hunderte Low-Priority-Test-Failures trainieren dein Team, Alerts zu ignorieren. Staffel deine Tests: P0 (blockt Pipeline), P1 (selber Tag fixen), P2 (Backlog).

3. Hardcoded Thresholds. "Row Count muss > 10.000 sein" bricht beim ersten langsamen Geschäftstag. Nutze stattdessen rollende Durchschnitte und prozentbasierte Schwellen.

4. Keine Quarantäne-Strategie. Wenn ein Test fehlschlägt — was passiert mit den Daten? Ohne Dead-Letter-Queue oder Quarantäne-Tabelle verschwinden gescheiterte Records einfach, und niemand merkt es.

5. Test-Wartung ignorieren. Quality-Tests brauchen Wartung wie jeder andere Code. Wenn Schemas sich entwickeln, müssen Tests nachziehen. Plane vierteljährliche Reviews deiner Test-Suite.

Ein Quality-Gate-Pattern implementieren

Das effektivste Pattern, das ich in Produktion gesehen habe: Quality-Tests als Pipeline-Gates behandeln, die Downstream-Processing blocken.

# Python — Quality gate pattern (framework-agnostic)

from dataclasses import dataclass
from enum import Enum

class Severity(Enum):
    CRITICAL = "critical"  # Blocks pipeline
    WARNING = "warning"    # Logs alert, continues
    INFO = "info"          # Logs only

@dataclass
class TestResult:
    name: str
    passed: bool
    severity: Severity
    details: str = ""

def run_quality_gate(results: list[TestResult]) -> bool:
    """Returns True if pipeline should proceed."""
    critical_failures = [
        r for r in results
        if not r.passed and r.severity == Severity.CRITICAL
    ]

    warnings = [
        r for r in results
        if not r.passed and r.severity == Severity.WARNING
    ]

    for w in warnings:
        print(f"WARNING: {w.name} — {w.details}")

    if critical_failures:
        for f in critical_failures:
            print(f"CRITICAL: {f.name} — {f.details}")
        print(f"Pipeline blocked: {len(critical_failures)} critical failures")
        return False

    print(f"Quality gate passed ({len(warnings)} warnings)")
    return True

# Example usage in an Airflow task or pipeline step:
# results = [
#     TestResult("pk_uniqueness", True, Severity.CRITICAL),
#     TestResult("freshness_check", False, Severity.WARNING, "Data is 3h old"),
#     TestResult("null_check_email", False, Severity.CRITICAL, "42 null emails"),
# ]
# if not run_quality_gate(results):
#     raise AirflowFailException("Quality gate failed")

Wo Harbinger Explorer reinpasst

Wenn du Quality auf Daten testest, die aus APIs gepullt werden — wo ein Großteil der Ingestion-Probleme entsteht — lässt Harbinger Explorer dich API-Responses direkt im Browser via DuckDB WASM abfragen. Du kannst SQL-Assertions gegen rohe API-Daten laufen lassen, bevor sie deine Pipeline erreichen, und Schema-Drift sowie Wert-Anomalien an der Quelle abfangen, ohne erst ein Ingestion-Skript zu schreiben.

FAQ

Wo sollte Data Quality Testing in der Pipeline starten?

So früh wie möglich: am Ingestion-Layer mit Schema-Validierung. Dort sind Tests am billigsten, und Probleme schlagen nicht auf nachgelagerte Tabellen durch.

Welches Tool ist am besten — dbt tests, Great Expectations oder Soda?

Bist du auf dbt, starte mit dbt-Tests. Brauchst du komplexere statistische Assertions, ergänze Great Expectations. Bist du nicht auf dbt, ist Soda meist die sanfteste Lernkurve.

Wie verhindere ich Alert Fatigue?

Tests in Severity-Tiers staffeln (P0/P1/P2), rollende statt fester Schwellen nutzen, und Warnings konsolidiert (nicht pro Failure) zustellen. Wenn dein On-Call-Engineer Alerts wegklickt, hast du das Tier-Modell falsch.

Wie testet man DSGVO-relevante Daten richtig?

PII-Felder bekommen zusätzliche Checks: Validität (E-Mail-Regex, Telefonformate), Vollständigkeits-Grenzen (kein Massen-Mismatch), und Quarantäne für ungültige Records — niemals einfach durchschreiben. Lösch-Pipelines bekommen Tests, die garantieren, dass right_to_be_forgotten-Requests komplett durchgelaufen sind.

Wie viel Test-Coverage ist genug?

Pragmatisch: für jede produktive Tabelle mindestens fünf Assertions (Uniqueness, Completeness, Freshness, Referenzielle Integrität, Volume). Für Metrik-Tabellen zusätzlich statistische Anomalie-Detection.

Nächste Schritte

Data Quality Testing ist kein einmaliges Projekt. Starte mit dem höchsten Impact-Layer: Schema-Validierung beim Ingest. Ergänze SQL-Assertions für deine kritischsten Tabellen. Bau statistische Checks für deine Key-Metriken. Vor allem: lass Quality-Gates Pipelines blockieren — nicht einfach Failures loggen und hoffen, dass jemand die Logs liest.

Dein konkreter nächster Schritt: such die drei business-kritischsten Tabellen aus und schreib pro Tabelle fünf Assertions (Uniqueness, Completeness, Freshness, Referenzielle Integrität, Volume). Deploy sie morgen. Du wirst besser schlafen.

Weiterlesen

• Was ist dbt? Ein Data Engineer's Guide
• ETL vs ELT: Welcher Ansatz passt zu deinem Stack?
• Data Testing Frameworks

Stand: 14. Mai 2026.