In der heutigen Datenwelt liegt der Fokus oft darauf, SQL-Abfragen zu schreiben, die „funktionieren“. Doch nur wenige testen, ob diese Abfragen auch morgen noch korrekt sind. Eine neue Zeile, eine geänderte Annahme oder eine Umstrukturierung können eine Abfrage stillschweigend zum Scheitern bringen. Dieser Artikel beschreibt einen vollständigen Workflow, der zeigt, wie SQL wie Software behandelt werden kann: versioniert, getestet und automatisiert. Wir verwenden eine reale Interviewfrage von Amazon, die darauf abzielt, die Kunden mit den höchsten täglichen Ausgaben zu identifizieren. Anschließend wandeln wir die SQL-Abfrage in eine testbare Komponente um, definieren die erwarteten Ausgaben und automatisieren die Tests mit kontinuierlicher Integration und kontinuierlicher Bereitstellung (CI/CD).
Schritt 1: Lösung einer Interview-ähnlichen SQL-Frage
In dieser Interviewfrage von Amazon wird verlangt, die Kunden mit den höchsten täglichen Gesamtausgaben innerhalb eines bestimmten Datumsbereichs zu finden.
Verständnis des Datensatzes
Für dieses Projekt stehen zwei Datentabellen zur Verfügung: Kunden und Bestellungen.
Die Tabelle der Kunden:
Hier ist eine Vorschau des Datensatzes:
Die Tabelle der Bestellungen:
Hier ist eine Vorschau des Datensatzes:
Dieses Problem eignet sich hervorragend, um zu veranschaulichen, wie SQL wie Software behandelt werden kann: Die Abfrage muss korrekt, stabil und resistent gegen Regressionen sein. Weitere Informationen dazu finden Sie in unserem Artikel über Wie die Struktur von Datensätzen den Programmierstil beeinflusst.
Schreiben der SQL-Lösung
Die Logik lässt sich in drei Teile unterteilen:
- Die Gesamtausgaben jedes Kunden pro Tag aggregieren
- Kunden nach Gesamtausgaben für jedes Datum ranken
- Nur die täglichen Spitzenausgeber zurückgeben
Hier ist die endgültige PostgreSQL-Lösung:
WITH customer_daily_totals AS (
SELECT
o.cust_id,
o.order_date,
SUM(o.total_order_cost) AS total_daily_cost
FROM orders o
WHERE o.order_date BETWEEN ‚2019-02-01‘ AND ‚2019-05-01‘
GROUP BY o.cust_id, o.order_date
),
ranked_daily_totals AS (
SELECT
cust_id,
order_date,
total_daily_cost,
RANK() OVER (
PARTITION BY order_date
ORDER BY total_daily_cost DESC
) AS rnk
FROM customer_daily_totals
)
SELECT
c.first_name,
rdt.order_date,
rdt.total_daily_cost AS max_cost
FROM ranked_daily_totals rdt
JOIN customers c ON rdt.cust_id = c.id
WHERE rdt.rnk = 1
ORDER BY rdt.order_date;
Definition der erwarteten Ausgabe
Hier ist die erwartete Ausgabe:
An diesem Punkt hören die meisten Menschen auf.
Schritt 2: Zuverlässigkeit der SQL-Logik durch Einheitstests
SQL bricht leichter, als viele denken. Eine geänderte Standardeinstellung, eine umbenannte Spalte oder eine neue Datenquelle können stille Fehler einführen. Tests schützen vor diesen Problemen. Es gibt drei Testschritte, die wir behandeln werden: die Logik in eine Funktion umwandeln, die erwartete Ausgabe definieren und eine Einheitstest-Suite schreiben. In Bezug auf die Automatisierung können Effiziente Python-Skripte zur Automatisierung der explorativen Datenanalyse hilfreich sein.
Umwandlung der Abfrage in eine wiederverwendbare Komponente
Um den SQL-Code zu testen, beginnen wir damit, ihn in eine Python-Funktion einzuwickeln, die ein leichtgewichtiges Testframework wie unittest verwendet. Zuerst definieren wir die Abfrage, die wir testen möchten:
query = \“\“\“
WITH customer_daily_totals AS (
SELECT
o.cust_id,
o.order_date,
SUM(o.total_order_cost) AS total_daily_cost
FROM orders o
WHERE o.order_date BETWEEN ‚2019-02-01‘ AND ‚2019-05-01‘
GROUP BY o.cust_id, o.order_date
),
ranked_daily_totals AS (
SELECT
cust_id,
order_date,
total_daily_cost,
RANK() OVER (
PARTITION BY order_date
ORDER BY total_daily_cost DESC
) AS rnk
FROM customer_daily_totals
)
SELECT
c.first_name,
rdt.order_date,
rdt.total_daily_cost AS max_cost
FROM ranked_daily_totals rdt
JOIN customers c ON rdt.cust_id = c.id
WHERE rdt.rnk = 1
ORDER BY rdt.order_date;
\“\“\“
Definition von Testdaten und erwarteter Ausgabe
Als Nächstes müssen wir einen kontrollierten Beispieldatensatz erstellen, gegen den wir testen können.
test_customers = [
(15, „Mia“),
(7, „Jill“),
(3, „Farida“)
]
test_orders = [
(1, 3, „2019-03-04“, 100),
(2, 3, „2019-03-01“, 80),
(4, 7, „2019-02-01“, 25),
(6, 15, „2019-02-01“, 100)
]
Wir erstellen auch die erwartete Ausgabe:
expected = [
(„Mia“, „2019-02-01“, 100),
(„Farida“, „2019-03-01“, 80),
(„Farida“, „2019-03-04“, 100)
]
Warum? Weil die Definition der erwarteten Ausgaben einen Maßstab schafft.
Schreiben von SQL-Einheitstests
Jetzt haben wir die Abfrage definiert, die Testdaten und die erwarteten Ausgaben. Wir können einen tatsächlichen Einheitstest schreiben. Die Idee ist einfach:
- Erstellen Sie eine isolierte, im Speicher befindliche Datenbank
- Laden Sie kontrollierte Testdaten
- Führen Sie die SQL-Abfrage aus
- Überprüfen Sie, ob das erhaltene Ergebnis mit der erwarteten Ausgabe übereinstimmt
Das integrierte unittest-Framework von Python ist äußerst effektiv, da es uns ermöglicht, die Abhängigkeiten minimal zu halten und gleichzeitig Struktur und Wiederholbarkeit zu bieten. Wir beginnen mit der Erstellung einer im Speicher befindlichen SQLite-Datenbank:
conn = sqlite3.connect(„:memory:“)
cursor = conn.cursor()
Die Verwendung von :memory: stellt sicher, dass:
- Die Testdatenbank vollständig isoliert ist
- Kein externer Zustand das Ergebnis beeinflussen kann
- Die Datenbank automatisch verworfen wird, sobald der Test abgeschlossen ist
Als Nächstes stellen wir nur die Tabellen wieder her, die für die Abfrage erforderlich sind:
CREATE TABLE customers (…)
CREATE TABLE orders (…)
Obwohl die Abfrage nur einen Teil der Spalten verwendet, spiegelt das Schema eine realistische Produktionstabelle wider. Dies verringert das Risiko von falschem Vertrauen, das durch vereinfachte Schemata verursacht wird. Dann fügen wir die zuvor definierten kontrollierten Testdaten ein:
cursor.executemany(„INSERT INTO customers VALUES (?, ?, ?, ?, ?, ?)“, test_customers)
cursor.executemany(„INSERT INTO orders VALUES (?, ?, ?, ?, ?)“, test_orders)
conn.commit()
Zu diesem Zeitpunkt enthält die Datenbank einen bekannten, deterministischen Zustand, der für sinnvolle Tests entscheidend ist. Bevor wir die Abfrage ausführen, laden und drucken wir die Testtabellen mit Pandas:
customers_df = pd.read_sql(„SELECT id, first_name, last_name, city FROM customers“, conn)
orders_df = pd.read_sql(„SELECT * FROM orders“, conn)
Obwohl dieser Schritt für die Automatisierung nicht unbedingt erforderlich ist, ist er während der Entwicklung und Fehlersuche äußerst nützlich. Wenn ein Test fehlschlägt, spart es erheblich mehr Zeit, die Eingabedaten sofort zu überprüfen, als die SQL-Logik zu überprüfen, da es ermöglicht, Schritt für Schritt zu verstehen, was der Code berechnet. Jetzt führen wir die zu testende Abfrage aus:
result = pd.read_sql(query, conn)
Das Ergebnis wird in ein DataFrame geladen, das Folgendes bietet:
- strukturierten Zugriff auf Zeilen und Spalten
- einfache Vergleichsmöglichkeiten mit den erwarteten Ausgaben
- lesbare Ausgabe für die Fehlersuche
Als Nächstes müssen wir die Ergebnisse zeilenweise überprüfen. Die Überprüfungslogik führt eine manuelle Überprüfung zwischen der Abfrageausgabe und dem erwarteten Ergebnis durch:
all_correct = True
if len(result) != len(expected):
all_correct = False
Die erste Überprüfung bestätigt, ob die Anzahl der von der Abfrage zurückgegebenen Zeilen mit dem übereinstimmt, was wir erwarten. Eine Abweichung hier deutet sofort auf fehlende oder zusätzliche Datensätze hin. Als Nächstes iterieren wir durch die erwartete Ausgabe und vergleichen sie zeilenweise mit dem tatsächlichen Abfrageergebnis:
for i, (fname, lname, date, cost) in enumerate(expected):
if i
Jede Zeile wird in allen relevanten Dimensionen überprüft:
- Kundenname
- Bestelldatum
- aggregierte tägliche Kosten
Wenn sich ein Wert von dem erwarteten unterscheidet, wird der Test als fehlgeschlagen markiert. Schließlich wird das Testergebnis in einer klaren Pass/Fail-Nachricht zusammengefasst:
if all_correct and len(result) == len(expected):
print(„ALLE TESTS BESTANDEN“)
else:
print(„EINIGE TESTS SIND FEHLGESCHLAGEN“)
Die Datenbankverbindung wird dann geschlossen:
conn.close()
Wenn die Tests bestehen, ist die erwartete Ausgabe:
Dieser Test bringt einige Annahmen mit sich, die beachtet werden sollten:
- eine stabile Zeilenreihenfolge (ORDER BY order_date)
- exakte Übereinstimmungen aller Werte
- keine Toleranz für Gleichstände oder doppelte Gewinner pro Tag
Das vollständige Skript, bereit zur Verwendung, kann hier eingesehen werden.
Schritt 3: Automatisierung von SQL-Tests mit kontinuierlicher Integration und kontinuierlicher Bereitstellung
Eine Test-Suite ist nur dann nützlich, wenn sie konsequent ausgeführt wird, wann immer es nötig ist. Wir nutzen CI/CD, um die Tests zu automatisieren, wann immer eine Codeänderung vorgenommen wird.
Organisation des Projekts
Eine minimale Repository-Struktur könnte folgendermaßen aussehen:
Erstellung des GitHub Actions Workflows
Der nächste Schritt besteht darin, sicherzustellen, dass diese Tests automatisch ausgeführt werden, wann immer sich der Code ändert. Dazu verwenden wir GitHub Actions. Dieses Tool ermöglicht es uns, einen CI-Workflow zu definieren, der die SQL-Tests jedes Mal ausführt, wenn Code gepusht oder ein Pull-Request geöffnet wird.
Erstellen Sie die Workflow-Datei: In Ihrem Repository erstellen Sie die folgende Ordnerstruktur, falls sie noch nicht vorhanden ist: .github/workflows/. Innerhalb dieses Ordners erstellen Sie eine neue Datei mit dem Namen test_sql.yml. Der Name ist nicht wichtig; GitHub interessiert sich nur dafür, dass die Datei im Verzeichnis .github/workflows/ vorhanden ist. Sie können sie beliebig benennen, aber test_sql.yml hält die Dinge klar und einfach.
Definieren, wann der Workflow ausgeführt werden soll
Hier ist die vollständige Workflow-Datei:
name: Run SQL Tests
on:
push:
branches: [ „main“ ]
pull_request:
branches: [ „main“ ]
Dieser Abschnitt definiert, wann der Workflow ausgeführt wird:
- bei jedem Push in den Hauptbranch
- bei jedem Pull-Request, der auf den Hauptbranch abzielt
In der Praxis bedeutet dies:
- Ein direktes Pushen in den Hauptbranch löst die Tests aus
- Das Öffnen oder Aktualisieren eines Pull-Requests löst ebenfalls die Tests aus
Dies hilft, SQL-Regressionen zu erkennen, bevor sie zusammengeführt werden.
Definieren des Testjobs
Als Nächstes definieren wir einen Job namens test:
jobs:
test:
runs-on: ubuntu-latest
Dies weist GitHub an:
- eine frische Linux-Maschine zu erstellen
- alle Testschritte darin auszuführen
Jeder Workflow-Lauf beginnt in einer sauberen Umgebung, was „Es funktioniert auf meiner Maschine“-Probleme verhindert.
Hinzufügen der Workflow-Schritte
Jetzt definieren wir die Schritte, die die Maschine ausführen soll:
– name: Checkout repository
uses: actions/checkout@v4
Dieser Schritt lädt den Code Ihres Repositories in den Runner, sodass er auf Ihre SQL-Dateien und Tests zugreifen kann.
– name: Set up Python
uses: actions/setup-python@v5
with:
python-version: „3.10“
Dies installiert Python 3.10 und stellt sicher, dass eine konsistente Laufzeit über alle Ausführungen hinweg gewährleistet ist.
– name: Install dependencies
run: |
python -m pip install –upgrade pip
pip install -r requirements.txt
Dies installiert alle erforderlichen Python-Bibliotheken (wie Pandas), die in requirements.txt definiert sind.
– name: Run unit tests
run: python -m unittest discover
Schließlich führt dieser Befehl:
- automatisch die Testdateien auf
- führt alle SQL-Tests aus, die im tests/-Ordner definiert sind
- lässt den Workflow fehlschlagen, wenn ein Test fehlschlägt
Der vollständige Workflow kann hier eingesehen werden.
Ausführen des Workflows
Sie müssen diese Datei nicht manuell ausführen. Sobald sie eingegeben ist:
- Ein Push in den Hauptbranch löst den Workflow aus
- Das Öffnen eines Pull-Requests löst den Workflow aus
Sie können die Ergebnisse direkt in GitHub anzeigen, indem Sie zum Tab „Aktionen“ Ihres Repositories navigieren.
Jeder Lauf zeigt, ob Ihre SQL-Tests bestanden oder fehlgeschlagen sind.
Schritt 4: Automatisierung der Datenqualität
Einheitstests bestätigen, ob die Logik weiterhin die erwartete Ausgabe zurückgibt, und CI stellt sicher, dass diese Tests automatisch ausgeführt werden. Doch in realen Datenumgebungen kann die Eingabedaten selbst zu Fehlern führen: verspätete Zeilen, fehlerhafte Daten, fehlende Schlüssel und unerwartete Duplikate können Abfragen lange vor der SQL-Logik zum Scheitern bringen. Hier kommt die Automatisierung der Datenqualität ins Spiel. Tests und Versionierung bilden ein Sicherheitsnetz für Codeänderungen; die Automatisierung der Datenqualität erweitert dieses Sicherheitsnetz auf die Daten selbst und verhindert nachgelagerte Probleme, bevor sie die Ergebnisse beeinträchtigen.
Verständnis, warum Datenqualitätsprüfungen für SQL-Workflows wichtig sind
In unserem Interviewproblem könnten folgende Probleme dazu führen, dass die Abfrage falsche Ergebnisse zurückgibt:
- Der Vorname eines Kunden ist nicht mehr eindeutig.
- Eine Bestellung kommt mit negativen Kosten an.
- Datumsangaben liegen außerhalb des erwarteten Bereichs.
- Tägliche Aggregationen enthalten doppelte Zeilen für denselben Kunden und dasselbe Datum.
- Ein Kunde existiert in Bestellungen, aber nicht in Kunden.
Ohne automatisierte Prüfungen können diese Probleme stillschweigend die Ergebnisse verzerren. Da SQL in vielen dieser Szenarien keine offensichtlichen Ausnahmen auslöst, breiten sich Fehler unbemerkt aus. Automatisierte Daten
„`
Bildquelle: ai-generated-gemini
