Data Mesh & Data Engineering – Portfolioprüfung (Gruppe 3)

Datenprodukt „Standortprofil deutscher Kreise" auf Basis von vier öffentlichen Datensätzen (Gemeinden, Baulandverkäufe, Klimadaten, Bevölkerungszahlen). Die Rohdaten liegen fertig in einer Cloudera CDP / Impala-Datenbank der DHBW Stuttgart (default.project_*, nur lesend); unser Datenprodukt entsteht in der Gruppen-Datenbank gruppe3.

Use Case: Landkreise vergleichbar machen (Wohnraumdruck, Baulandpreise, Klimarisiko), indem Bevölkerung + Bauland + Klima + Gemeinden zu einem analysefertigen Datenprodukt verbunden werden — Zielbild ist ein Standortprofil-Dashboard.

Die Pipeline folgt dem Write-Audit-Publish-Pattern (WAP) in drei Stufen und lädt inkrementell (Wasserzeichen, zeilengenauer Key-Merge via Apache Iceberg, Inhalts-Prüfsummen — unveränderte Quellen werden übersprungen). Die eigentlichen Transformationen (Unpivot, Pivot, Window-Funktionen) laufen in Apache Spark (PySpark, DataFrame-API).

Doku-Struktur (genau 3 Markdown-Dateien):

Datei	Frage, die sie beantwortet
README.md (diese Datei)	Was ist das Projekt, wie funktioniert es, wie führe ich es aus, warum sieht das Datenmodell so aus?
docs/ADR.md	Warum ist alles so gebaut — alle Architektur-Entscheidungen (ADR-1..17) + abgelöste Entscheidungen
docs/Probleme.md	Alle Probleme: gelöst (mit Fallstudien), nicht behebbar (dokumentiert), offen

Die vier abgegebenen Arbeitsergebnisse:

Deliverable	Wo
1. DDLs + Begründung des Datenmodells	src/create_datamodel.py + Datenmodell & Begründung (unten)
2. Pipeline-Code (3 Stufen + Orchestrator + Incremental-State) + Scheduler	src/run_pipeline.py, src/pipeline_default_to_staging.py, src/pipeline_staging_to_audit.py, src/pipeline_audit_to_target.py, src/etl_state.py, src/scheduler.py
3. Data Contract + technische Durchsetzung	data/data_contract.yaml, data/output_port_ddl.sql, src/contract_check.py
4. README (dieses Dokument)	–

Inhalt: Architektur · Datenmodell · Incremental Loading · Projektstruktur · Lese-Reihenfolge · Einrichtung · Benutzung · Docker · Betriebsarten & CDE · Datenbank & Datenprodukt · Output Port · Datenqualität · Prüfungswissen · Stand

Architektur: der Datenfluss

default.project_*          (Rohdaten, 4 Tabellen, nur lesend; Klima: 8,6 Mio. Zeilen)
      │
      │  (1) pipeline_default_to_staging.py      WRITE   – impyla / Impala-SQL
      ▼       Klimadaten: Wasserzeichen-Append (nur neue Tage) · übrige Tabellen:
              Prüfsummen-Check, Full Refresh nur bei echter Änderung
gruppe3_staging_*          (unveränderte Rohkopie in unserer Datenbank)
      │
      │  (2) pipeline_staging_to_audit.py        AUDIT   – impyla / Impala-SQL
      ▼       Bereinigung (Encoding-Korrekturen, Transliteration, Koordinaten,
              Filter Germany); inkrementell: Wasserzeichen (Klima) · zeilengenauer
              Business-Key-Merge via Iceberg MERGE/DELETE (Bauland/Bevölkerung)
              · Prüfsumme (Gemeinden)
gruppe3_audit_*            (bereinigte, fachlich saubere Basis)
      │
      │  (3) pipeline_audit_to_target.py         PUBLISH – Apache Spark (PySpark)
      ▼       Unpivot (Bevölkerung), Pivot (Bauland), Jahresmittel (Klima),
              Namens-Matching, Window-Funktionen/z-Score → Star-Schema;
              der komplette Spark-Lauf wird übersprungen, wenn sich seit dem
              letzten Ziel-Build keine Audit-Tabelle geändert hat
gruppe3_dim_* / gruppe3_fact_*   (das Datenprodukt, s. Data Contract)
      │
      │  (4) contract_check.py                   GATE    – impyla / Data Contract
      ▼       Schema, Pflichtfelder, Eindeutigkeit und ausführbare Quality-SQLs;
              bricht den Lauf mit Exit-Code 1 ab, wenn der Contract verletzt ist.

src/run_pipeline.py führt alles in fester Reihenfolge aus (Datenmodell-DDLs + Stufen 1–3 + Data-Contract-Gate) und ist der eine Einstiegspunkt — auch im Docker-Image.

Warum Stufe 1+2 in Impala-SQL und nur Stufe 3 in Spark? Kopieren und zeilenweises Bereinigen sind reine INSERT OVERWRITE/INTO … SELECT-Operationen — die laufen serverseitig in Impala, ohne 8,6 Mio. Zeilen durch Python/Spark zu schleusen. Spark wird dort eingesetzt, wo es echten Mehrwert hat: Unpivot per explode(array(struct(…))), echtes DataFrame.pivot() und Window-Aggregate (STDDEV() OVER für den z-Score — in Impala-SQL nicht möglich).

Wie die Pipeline dabei unveränderte Quellen erkennt und überspringt, erklärt der Abschnitt Incremental Loading — er folgt direkt auf das Datenmodell: erst das Ziel, dann die Lade-Mechanik.

Datenmodell & Begründung

Ausgangslage: vier Quelltabellen

Quelle (`default.*`)	Ebene	Verknüpfungs-Schlüssel	Format / Besonderheit
`project_bevoelkerungzahlen`	Kreis	`id` (Regionalschlüssel)	breit (92 Spalten: 3 je Jahr 1995–2024); kaputte Umlaute (`�`)
`project_bauland`	Kreis	`kreis_id` (Regionalschlüssel)	lang (1 Zeile je Merkmal, 21.600 Zeilen); kaputte Umlaute; 4. Merkmal ab Quelle zerstört
`project_gemeinden`	Gemeinde	nur Name (kein Schlüssel), `latitude`/`longitude`	CSV-Parsing-Schäden, 10.950 Zeilen
`project_klimadaten`	Stadt (weltweit)	nur Stadtname, `latitude`/`longitude`	8,6 Mio. Zeilen, endet 2013; Kompass-Koordinaten („53.84N")

Zentrale Erkenntnis 1: bevoelkerung.id und bauland.kreis_id sind derselbe amtliche Regionalschlüssel (z. B. 01001 = Flensburg). Sie matchen exakt (472 Kreise, live geprüft) und sind hierarchisch: die ersten 2 Stellen kodieren das Bundesland (01 = Schleswig-Holstein). → Der natürliche Integrationsschlüssel auf Kreis-Ebene. (Im fertigen dim_kreis stehen 474 Kreise: Berlin und Hamburg führen in der Quelle nur ihren 2-stelligen Landesschlüssel und werden erst in Stufe 3 auf 11000/02000 gehoben — s. Datenqualität.)

Zentrale Erkenntnis 2: project_gemeinden und project_klimadaten lassen sich über die Gemeinde-Ebene verbinden: Nach der Bereinigung in Stufe 2 (Transliteration ä→ae/…, englische Exonyme „Munich"→„Muenchen" gemappt) liegen Gemeindenamen und Klimastadt-Namen im selben Format vor → exakter Namens-Match (deterministisch, trifft 74 von 81 deutschen Klimastädten). dim_gemeinde wird damit zur Brücken-Dimension, die Kreis-Ebene (Bevölkerung/Bauland) und Stadt-Ebene (Klima) verbindet — alle vier Quellen sind in einem Modell nutzbar, nicht nur isoliert nebeneinander.

Gewähltes Modell: Star-/Galaxy-Schema (4 Dimensionen + 5 Fakten)

dim_kreis ──< fact_bevoelkerung   >── dim_jahr
dim_kreis ──< fact_bauland        >── dim_jahr
dim_kreis ──< dim_gemeinde ──< fact_gemeinde_stamm
dim_gemeinde ──(Namens-Match, transliteriert)──> dim_klimastadt ──< fact_klima >── dim_jahr
dim_kreis ──< fact_standortprofil_kpi >── dim_jahr   (verdichtet alle o.g. Fakten)

Genau genommen ein Galaxy-Schema: mehrere Faktentabellen teilen sich die Dimensionen dim_kreis und dim_jahr (conformed dimensions); jede Faktentabelle + ihre Dimensionen = ein Stern.

Warum so? (die Begründung)

Star-Schema statt normalisiert (3. NF). Analytische (OLAP-)Systeme lesen Aggregate über viele Zeilen; normalisierte Modelle erzwingen viele Joins → komplexe, langsame, in der Cloud teure Abfragen. Die Dimensionen sind denormalisiert (z. B. Bundesland direkt in dim_kreis).
Regionalschlüssel als conformed dimension. dim_kreis verbindet fact_bevoelkerung und fact_bauland exakt — Data-Mesh-Gedanke: Mehrwert durch Interconnecting.
dim_gemeinde als zweite conformed dimension (Brücke). Ohne sie bliebe das Klima isoliert (kein Regionalschlüssel): Namens-Match gegen dim_kreis.kreis_name (Kreis-Anbindung) und gegen dim_klimastadt.stadt_name (Klima-Anbindung).
Unpivot der Bevölkerungsdaten (breit → lang). Aus 1 Spalte pro Jahr wird 1 Zeile je Kreis+Jahr — jahr wird echte Dimension, Zeitreihen-Analysen trivial. In Spark per explode(array(struct(…))).
Pivot der Baulanddaten (lang → breit). Aus 1 Zeile je Merkmal wird 1 Spalte je Merkmal (alle 4: Fälle, Fläche, Kaufsumme, amtl. Kaufwert) — genau eine Faktenzeile je Kreis+Jahr. In Spark per DataFrame.pivot().
KPI-Spalten direkt in den Basisfakten, wo sie aus derselben Zeile berechenbar sind (z. B. preis_pro_qm_eur) — statt bei jeder Abfrage neu.
fact_standortprofil_kpi als aggregierter Cross-Table-Fakt. Kennzahlen, die den Join mehrerer Fakten erfordern (wohnraumdruck_index, standortattraktivitaets_score), werden einmal in der Pipeline vorberechnet und dashboard-fertig auf Kreis × Jahr gespeichert.
Parquet (spaltenorientiert) für alle Zieltabellen → ideal für OLAP („data skipping", nur benötigte Spalten werden gelesen).
Bewusste Abgrenzung: Die Namens-Matches sind fehlerbehaftet und ehrlich im Data Contract dokumentiert (97,5 % Kreis-Coverage, 74/81 Klimastädte) — Data-Mesh-Prinzip „Federated Governance": Qualität beschreiben statt verschweigen.

Idempotenz

Alle DDLs nutzen CREATE TABLE IF NOT EXISTS; die Befüllung ist inkrementell (Wasserzeichen-Append, Key-Merge bzw. INSERT OVERWRITE nur bei echter Änderung). Mehrfaches Ausführen erzeugt keine Duplikate — ein zweiter Lauf direkt nach dem ersten meldet überall „übersprungen" (Stufe 1+2 live verifiziert am 10.07.2026; der komplette Zwei-Läufe-Test inkl. Stufe 3 steht noch aus, s. Stand).

Incremental Loading: wie die Pipeline Änderungen erkennt

Den zuletzt verarbeiteten Stand merkt sich die Pipeline in gruppe3_etl_state (Iceberg, ein Eintrag je Stufe+Tabelle, per MERGE INTO geupsertet; die Verlaufs-Historie liegt in den Iceberg-Snapshots — s. src/etl_state.py).

Ebene 1 — Ganztabellen-Fingerprint (ohne Zeilen-Scan): Vor jeder Stufe wird zuerst EIN gespeicherter Fingerprint je Tabelle verglichen — für Iceberg-Tabellen die aktuelle Snapshot-ID (DESCRIBE HISTORY, reine Metadaten-Abfrage). Unverändert → die Stufe wird komplett übersprungen, ohne eine einzige Datenzeile zu lesen. Erst bei einem Fingerprint-Wechsel greift Ebene 2, der eigentliche Incremental-Mechanismus. Drei Strategien, je nach Tabellenart:

Wasserzeichen (Klimadaten, echte Zeitreihe über dt): nur neuere Zeilen werden angehängt — kein täglicher Full-Rewrite von 8,6 Mio. Zeilen. Die Iceberg-Partitionierung TRUNCATE(4, dt) (= Jahr) macht den Wasserzeichen-Filter zusätzlich zum Partition-Prune.
Zeilengenauer Merge (Bauland, Bevölkerung — Tabellen mit Business-Key): billiger Prüfsummen-Vor-Check; bei Änderung direktes Iceberg DELETE + MERGE INTO gegen die Audit-Tabelle mit NULL-sicherem Spaltenvergleich (t.col <=> src.col) — nur tatsächlich neue/geänderte/ gelöschte Zeilen werden angefasst, die Audit-Tabelle selbst ist der Vergleichsstand (keine separate Zeilen-Hash-Historie nötig).
Inhalts-Prüfsumme (Gemeinden — kein verlässlicher Key): die ganze Tabelle wird serverseitig gehasht (FNV_HASH je Zeile + SUM) und mit dem gespeicherten Hash verglichen. Unverändert → Schritt wird übersprungen; geändert → Full Refresh per INSERT OVERWRITE (auf Iceberg ein einzelner atomarer Snapshot-Commit).

Der Skip-Mechanismus zieht sich durch die ganze Pipeline: Stufe 3 überspringt den kompletten Spark-Lauf, wenn jede Audit-Tabelle noch exakt die Iceberg-Snapshot-ID hat, die der letzte Ziel-Build gelesen hat (Vergleich des Datenstands statt eines Zeitstempels), und der Publish selbst ist ein atomarer Shadow-Swap (s. overwrite_table in src/pipeline_audit_to_target.py). Einzige Stelle, an der im Unverändert-Fall noch Zeilen gescannt werden: die fremden Quelltabellen in default.* (kein Iceberg → kein verlässlicher Metadaten-Fingerprint → Prüfsummen-Scan als Fallback).

Die fachlichen Begründungen stehen ausführlich in den Modul-Docstrings der jeweiligen Skripte; die Iceberg-Entscheidung inkl. Live-Verifikation in docs/ADR.md (ADR-13).

Projektstruktur

Data-Mesh/
├── README.md                        # Diese Datei
├── requirements.txt                 # Python-Abhängigkeiten (impyla, pyspark, APScheduler, dotenv, PyYAML)
├── .env.example                     # Vorlage für Zugangsdaten → kopieren nach .env
├── .env                             # Echte Zugangsdaten (NICHT eingecheckt)
├── .gitignore / .dockerignore
├── Dockerfile                       # Container-Image (Python 3.11 + JDK 17)
├── docker-compose.yml               # Dienste: pipeline (Komplett-Lauf) + scheduler (dauerhaft)
│
├── src/
│   ├── db.py                        # Zentraler Impala-Verbindungs-Helfer (impyla + .env)
│   ├── create_datamodel.py          # DELIVERABLE 1: DDLs Star-Schema (4 Dim + 5 Fakten), idempotent
│   ├── run_pipeline.py              # DELIVERABLE 2: Orchestrator – Datenmodell + alle 3 Stufen + Contract-Gate
│   ├── pipeline_default_to_staging.py   # Stufe 1: Rohdaten → Staging (inkrementell)
│   ├── pipeline_staging_to_audit.py     # Stufe 2: Staging → Audit (Bereinigung, inkrementell)
│   ├── pipeline_audit_to_target.py      # Stufe 3: Audit → Datenmodell (Spark, mit Skip-Check)
│   ├── contract_check.py            # DELIVERABLE 3b: Data Contract als technisches Publish-Gate
│   ├── etl_state.py                 # Incremental-Loading-Zustand (Iceberg-Upsert) + Iceberg-Helfer
│   ├── scheduler.py                 # Täglicher Batch-Lauf um 00:00 (APScheduler)
│   └── utils/
│       ├── reset_database.py        # Löscht ALLE gruppe3-Tabellen (Reset für End-to-End-Tests, fragt nach)
│       ├── german_cities.txt        # Referenzliste Städte/Orte (für Encoding-Auflösung in Stufe 2)
│       ├── german_regions.txt       # Referenzliste Landkreise + kreisfreie Städte
│       ├── german_states.txt        # Referenzliste der 16 Bundesländer
│       └── ImpalaJDBC42.jar         # JDBC-Treiber für Spark (lokal bereitzustellen, s. Einrichtung)
│
├── cde/
│   └── pipeline_dag.py              # Airflow-DAG für Cloudera Data Engineering (5 Stufen als Tasks; live seit 11.07.2026)
│
├── data/
│   ├── data_contract.yaml           # DELIVERABLE 3: Data Contract (Schema, Nutzung, Qualität)
│   └── output_port_ddl.sql          # SQL-Basis für datacontract import sql
│
└── docs/
    ├── ADR.md                       # ALLE Architektur-Entscheidungen (ADR-1..17) + abgelöste
    ├── Probleme.md                  # ALLE Probleme: gelöst / nicht behebbar / offen
    ├── pictures/                    # Architektur-Diagramme (drawio: WAP-Pipeline, Star-Schema, Contract-Gate)
    └── portfolio/Portfolioprüfung.pdf   # Aufgabenstellung

Wie liest man den Code?

Empfohlene Lese-Reihenfolge (jede Datei erklärt ihre Entscheidungen selbst im Modul-Docstring am Dateianfang):

src/db.py – wie die Verbindung zu Impala aufgebaut wird (alle Skripte nutzen diesen einen Helfer; Zugangsdaten kommen aus .env).
src/run_pipeline.py – der rote Faden: ruft Datenmodell + die drei Stufen + Contract-Gate in der richtigen Reihenfolge auf.
src/create_datamodel.py – die Zieltabellen mit COMMENT an jeder Spalte. Warum das Modell so aussieht: s. unten Datenmodell & Begründung.
src/pipeline_default_to_staging.py – Stufe 1: exakte Kopie (Schema per Iceberg-CTAS WHERE 1=0), inkl. der Entscheidung, welche Tabelle Wasserzeichen bekommt und welche Prüfsumme.
src/etl_state.py – das „Gedächtnis" der Pipeline: der Iceberg-Upsert-State (ein Eintrag je Stufe+Tabelle, Historie über Iceberg-Snapshots) und die gemeinsamen Iceberg-Helfer aller Stufen.
src/pipeline_staging_to_audit.py – Stufe 2, die Datenbereinigung. Hier steckt die Lösung des Encoding-Problems (Laufzeit-Erkennung kaputter Kreisnamen + automatische Auflösung über die Referenzlisten in src/utils/, Transliteration für intakte Umlaute, Koordinaten-Normalisierung) und der zeilengenaue Key-Merge per Iceberg MERGE INTO/DELETE (Hintergrund: ADR-13).
src/pipeline_audit_to_target.py – Stufe 3 (Spark): pro Zieltabelle eine build_…()-Funktion, main() führt sie in Abhängigkeits-Reihenfolge aus (erst Dimensionen, dann Basis-Fakten, zuletzt der aggregierte KPI-Fakt).
data/data_contract.yaml – der eigentliche Data Contract für den Output Port, kompatibel mit der Data Contract CLI. data/output_port_ddl.sql ist die SQL-Basis, aus der der Contract per CLI generiert bzw. abgeglichen werden kann.
src/contract_check.py – zusätzliches lokales Publish-Gate: prüft denselben Contract live gegen Impala und bricht bei Verstoß mit Exit-Code 1 ab.
src/scheduler.py – täglicher Batch-Trigger um 00:00.

Einrichtung (einmalig)

# 1. Virtuelle Umgebung + Abhängigkeiten
python -m venv .venv
.venv/Scripts/python.exe -m pip install -r requirements.txt

# 2. Zugangsdaten: .env.example nach .env kopieren und ausfüllen
#    (Workload-Username & -Passwort aus dem Cloudera-Portal)

Optionale Konfiguration (Standard reicht für die Abgabe): Die Ziel-Datenbank, das Tabellen-Präfix und die Quell-Datenbank sind über die .env überschreibbar (DATABASE=gruppe3, PREFIX=gruppe3_, SOURCE_DATABASE=default, s. ADR-14). Ausnahme: utils/reset_database.py ist bewusst fest auf gruppe3 verdrahtet (Schutz gegen versehentliches DROP TABLE in einer fremden DB).

Zusätzlich für Stufe 3 / run_pipeline.py / Scheduler (Spark):

JDK 17 installieren (genau Version 17 — neuere JDKs ≥ 23/24 entfernen Security-APIs, die Sparks Hadoop noch braucht, s. Probleme.md); Pfad in .env als JAVA_HOME_JDK17 eintragen. Die Skripte setzen JAVA_HOME daraus selbst (ADR-16).
src/utils/ImpalaJDBC42.jar: proprietärer Cloudera-Treiber, deshalb nicht eingecheckt (.gitignore) — die Datei muss lokal unter src/utils/ liegen (aus dem Cloudera-Portal laden bzw. im Team weitergeben).

Die Stufen 1+2 einzeln und alle utils/-Skripte brauchen kein Java — nur Python + .env.

Parquet→Iceberg-Migration (erledigt, historisch): Die einmalige Migration aller gruppe3-Bestandstabellen zu Iceberg wurde am 10.07.2026 gegen die echte Gruppen-Datenbank ausgeführt (Zeilenzahl-Verifikation vor jedem Tausch); das Migrationsskript wurde danach aus dem Repo entfernt. Die Pipeline bricht mit einer klaren Fehlermeldung ab, falls sie doch auf eine nicht migrierte Tabelle trifft. Bei einer frisch zurückgesetzten Datenbank ist keine Migration nötig — alle Tabellen werden direkt als Iceberg angelegt.

Benutzung

# KOMPLETTER LAUF (empfohlen): Datenmodell + Stufe 1 → 2 → 3 + Contract-Gate
.venv/Scripts/python.exe src/run_pipeline.py

Alle Schritte sind idempotent: beliebig oft ausführbar, keine Duplikate. Unveränderte Quellen werden erkannt und übersprungen — ein zweiter Lauf direkt nach dem ersten tut fast nichts (Stufe 1+2 live verifiziert, s. Stand).

Die Stufen lassen sich auch einzeln ausführen (gleiche Reihenfolge):

.venv/Scripts/python.exe src/create_datamodel.py                # Zieltabellen anlegen
.venv/Scripts/python.exe src/pipeline_default_to_staging.py     # Stufe 1
.venv/Scripts/python.exe src/pipeline_staging_to_audit.py       # Stufe 2
.venv/Scripts/python.exe src/pipeline_audit_to_target.py        # Stufe 3 (braucht JDK 17 + Treiber)
.venv/Scripts/python.exe src/contract_check.py                  # Data-Contract-Gate gegen das Datenprodukt

# Target-Rebuild erzwingen, wenn sich nur Code/Contract geändert hat, aber keine Audit-Daten
$env:FORCE_TARGET_BUILD="1"; .venv/Scripts/python.exe src/pipeline_audit_to_target.py; Remove-Item Env:FORCE_TARGET_BUILD

# Täglicher Lauf um 00:00 – läuft dauerhaft (lokale Betriebsart; nicht parallel zum CDE-DAG)
.venv/Scripts/python.exe src/scheduler.py

# Kompletter Reset der gruppe3-Tabellen (z.B. um Full Load vs. Skip zu testen)
.venv/Scripts/python.exe src/utils/reset_database.py

Danach ist das Datenprodukt abfragbar. Beispiel-Queries, Spaltenbedeutungen und NULL-Semantik stehen im Data Contract. Der letzte Schritt des Komplettlaufs erzwingt diesen Contract technisch: src/contract_check.py liest das YAML als Quelle der Wahrheit und prüft das veröffentlichte Datenprodukt live in Impala (aktuell 32 Checks).

Data Contract CLI

Der benotete Data Contract ist data/data_contract.yaml. Die zusätzliche Datei data/output_port_ddl.sql dokumentiert das physische Output-Port-Schema und kann mit der Data Contract CLI als Generierungsbasis genutzt werden:

python -m pip install "datacontract-cli[impala]" packaging

# Syntax/Schema des Contracts prüfen
$env:PYTHONIOENCODING="utf-8"; datacontract lint data/data_contract.yaml

# Contract aus den SQL-DDLs neu generieren/gegenprüfen
datacontract import sql --source data/output_port_ddl.sql --dialect spark --output data/data_contract.generated.yaml

# Contract gegen den Impala-Output-Port testen
# Vorher in .env pflegen: DATACONTRACT_IMPALA_USERNAME, DATACONTRACT_IMPALA_PASSWORD,
# DATACONTRACT_IMPALA_USE_SSL sowie im Contract host/port/database.
$env:PYTHONIOENCODING="utf-8"; datacontract test --server production data/data_contract.yaml

Falls datacontract test in der DHBW-Umgebung mit TSocket read 0 bytes abbricht, liegt das am Impala-Transport der CLI: die Pipeline nutzt IMPALA_HTTP_PATH/HTTP-Transport über impyla, der CLI-Adapter dagegen nur host, port, database, Username und Passwort. In diesem Fall bleiben datacontract lint und datacontract import sql gültig; das Live-Gate gegen den konkreten DHBW-Output-Port läuft über src/contract_check.py — kein Ersatz für die CLI, sondern ein zusätzliches Pipeline-Gate, weil die Pipeline bereits über impyla gegen dieselbe Impala-Umgebung läuft.

Handgeschriebener Contract vs. CLI-generiertes Gerüst

Zur Einordnung, was die CLI beiträgt (und was nicht), wurde der Contract am 09.07.2026 testweise per datacontract import sql (CLI 1.0.10) aus data/output_port_ddl.sql neu generiert. Ergebnis: das Schema stimmt 1:1 mit unserem Contract überein (9 Tabellen, alle Spalten und Typen) — mehr kann der Import aber prinzipbedingt nicht liefern, denn mehr steht nicht in einer DDL. Dieselbe Spalte im direkten Vergleich:

# CLI-generiert (nur, was aus der DDL ableitbar ist):
- name: kreis_id
  physicalType: STRING
  logicalType: string

# data/data_contract.yaml (fachlich angereichert):
kreis_id:
  type: string
  required: true      # nie NULL
  unique: true        # keine Duplikate
  primaryKey: true
  description: Amtlicher Regionalschlüssel, 5-stellig, z.B. '01001' (Flensburg).

Bestandteil	CLI-Import	Manuell ergänzt
Tabellen, Spalten, Typen	✓ automatisch, tippfehlerfrei aus der DDL	–
Constraints (`required`/`primaryKey`/`unique`)	–	✓
Spalten-Semantik inkl. Warnungen (z. B. `kaufwert_je_qm_eur`: „NICHT VERWENDEN")	–	✓
`terms` (Join-Regeln, Transliteration, Surrogat-Warnung)	–	✓
`servers` (erst damit ist `datacontract test` möglich)	–	✓
`quality` (ausführbare SQLs + live gemessene Zahlen)	–	✓
`examples`, `servicelevels`, Owner/Kontakt	–	✓

Was der CLI-Workflow bringt: (1) das mechanische Gerüst entsteht in Sekunden und ist per Konstruktion fehlerfrei, weil die DDL die Quelle der Wahrheit ist; (2) Drift-Erkennung — nach einer Schema-Änderung neu generieren und gegen den gepflegten Contract diffen, statt still zu veralten; (3) das Gerüst ist ab der ersten Zeile spezifikationskonform. Die Arbeitsteilung ist genau der empfohlene Weg „automatisiert erstellen, manuell ergänzen": die CLI liefert das Was (Schema), das Team den fachlichen Gehalt (Wie nutzt man es korrekt, was ist garantiert, was ist kaputt) — unser Contract entspricht dem Endzustand dieses Workflows.

Hinweis: Die CLI exportiert beim Import standardmäßig das ODCS-Format (Open Data Contract Standard, kind: DataContract/schema:) — der zweite große Contract-Standard neben der hier genutzten Data Contract Specification 1.1.0 (models:/fields:). Beide Standards konvergieren; datacontract lint/test verstehen beide.

Docker

Der Container enthält Python, PySpark und JDK 17; die Impala-Datenbank bleibt extern (DHBW), Zugangsdaten kommen zur Laufzeit aus .env (nicht im Image).

docker compose build

# Kompletter Pipeline-Lauf (run_pipeline.py, alle Stufen):
docker compose run --rm pipeline

# Dauerhafter lokaler Scheduler (nur, wenn der CDE-DAG nicht aktiv ist — s. „Zwei Betriebsarten"):
docker compose up scheduler

Wichtig — immer nur einen Dienst starten: kein nacktes docker compose up (würde pipeline und scheduler gleichzeitig starten; beide schreiben in dieselben Tabellen und kämen sich ins Gehege).

Zwei Betriebsarten: lokal (Docker) und Cloudera Data Engineering

Die Pipeline läuft mit demselben Code unter src/ in zwei Umgebungen:

Lokal / Docker (s. Benutzung und Docker oben): docker compose run --rm pipeline bzw. die .venv-Aufrufe dort. Spark läuft außerhalb des Clusters und erreicht die Daten über den Impala-JDBC-Endpoint (SPARK_IO_MODE=jdbc, Default — braucht das JDBC-Jar, s. Einrichtung).
CDE / Airflow (produktiver Scheduler — live seit 11.07.2026): Der Airflow-DAG gruppe3_data_mesh_pipeline orchestriert die fünf Stufen als CDE-Jobs und läuft täglich 05:00 UTC im Virtual Cluster (Monitoring, Retries, Wiederaufsetzen ab der gescheiterten Stufe in der Airflow-UI); Stufe 3 liest/schreibt Iceberg dort nativ über den Katalog (SPARK_IO_MODE=catalog — kein JDBC-Jar, kein collect()-Umweg). Setup, Deploy-Hinweise und Abnahme-Checkliste: s. direkt unten.

Beide Welten teilen sich gruppe3_etl_state (Fingerprint-Skip funktioniert über die Grenze hinweg); nur nicht beide Scheduler gleichzeitig betreiben: seit der DAG aktiv ist, den lokalen scheduler-Container nicht mehr starten — manuelle lokale Läufe (docker compose run --rm pipeline) bleiben dank Idempotenz jederzeit möglich.

	Lokal (Docker)	CDE
Taktgeber	`docker compose run pipeline` bzw. `scheduler.py`	Airflow-DAG (cde/pipeline_dag.py)
Stufen 0/1/2/4	impyla gegen Impala-Endpoint	identisch (impyla gegen Impala-Endpoint)
Stufe 3 Spark-I/O	`SPARK_IO_MODE=jdbc` (Default): JDBC-Jar + collect/VALUES/Shadow	`SPARK_IO_MODE=catalog`: native Iceberg-Reads/-Writes, kein Treiber-Jar
Zugangsdaten	`.env`	Env-Vars in der Job-Config (s. Deployment, Schritt 3)
ETL-State / Fingerprint-Skip	`gruppe3_etl_state`	dieselbe Tabelle — beide Welten teilen sich den Zustand

CDE-Deployment (einmalig, CDE CLI)

Voraussetzung: die CDE CLI ist gegen den Virtual Cluster der Gruppe konfiguriert; alle Kommandos aus dem Repo-Root (Data-Mesh/) ausführen. Dieses Setup ist am 11.07.2026 genau so ausgeführt worden — die Pipeline läuft seitdem im VC; die Schritte bleiben zur Reproduktion dokumentiert.

1. Code als Files-Resource hochladen. Die Verzeichnisstruktur muss erhalten bleiben (contract_check.py erwartet ../data/data_contract.yaml relativ zu src/):

cde resource create --name gruppe3-pipeline-code --type files
cde resource upload --name gruppe3-pipeline-code \
  --local-path src --resource-path src
cde resource upload --name gruppe3-pipeline-code \
  --local-path data/data_contract.yaml --resource-path data/data_contract.yaml

Das JDBC-Jar (src/utils/ImpalaJDBC42.jar) wird nicht gebraucht und nicht hochgeladen — im Katalog-Modus liest Spark den Metastore direkt.

2. Python-Environment anlegen. Die CDE-Jobs brauchen nur drei Pakete — ohne pyspark (stellt die CDE-Spark-Runtime selbst) und ohne APScheduler (Scheduling übernimmt der Airflow-DAG):

printf 'impyla==0.24.0\npython-dotenv==1.2.2\nPyYAML==6.0.2\n' > requirements-cde.txt

cde resource create --name gruppe3-python-env --type python-env
cde resource upload --name gruppe3-python-env \
  --local-path requirements-cde.txt --resource-path requirements.txt

(Danach den Build-Status abwarten: cde resource describe --name gruppe3-python-env.)

3. Die fünf Stufen-Jobs anlegen. Alle Stufen laufen als Spark-Jobs mit den vorhandenen Modulen als Application-File — die Stufen 0/1/2/4 nutzen Spark dabei gar nicht (reine Driver-Python-Skripte mit impyla), deshalb bekommen sie minimale Ressourcen. Die Zugangsdaten gehen als Driver-Env-Vars in die Job-Config (spark.kubernetes.driverEnv.* — CDE führt den Driver als Kubernetes-Pod aus):

# Gemeinsame Env-Vars fuer alle Jobs (Werte aus eurer .env):
ENVS="--conf spark.kubernetes.driverEnv.IMPALA_HOST=<host> \
 --conf spark.kubernetes.driverEnv.IMPALA_PORT=443 \
 --conf spark.kubernetes.driverEnv.IMPALA_HTTP_PATH=<http-path> \
 --conf spark.kubernetes.driverEnv.IMPALA_USER=<workload-user> \
 --conf spark.kubernetes.driverEnv.IMPALA_PASSWORD=<workload-passwort> \
 --conf spark.kubernetes.driverEnv.DATABASE=gruppe3 \
 --conf spark.kubernetes.driverEnv.PREFIX=gruppe3_"

cde job create --name gruppe3-stufe0-datenmodell --type spark \
  --application-file src/create_datamodel.py \
  --mount-1-resource gruppe3-pipeline-code \
  --python-env-resource-name gruppe3-python-env \
  --driver-cores 1 --driver-memory 1g --executor-cores 1 --executor-memory 1g --num-executors 1 \
  $ENVS

cde job create --name gruppe3-stufe1-staging --type spark \
  --application-file src/pipeline_default_to_staging.py \
  --mount-1-resource gruppe3-pipeline-code \
  --python-env-resource-name gruppe3-python-env \
  --driver-cores 1 --driver-memory 1g --executor-cores 1 --executor-memory 1g --num-executors 1 \
  $ENVS

cde job create --name gruppe3-stufe2-audit --type spark \
  --application-file src/pipeline_staging_to_audit.py \
  --mount-1-resource gruppe3-pipeline-code \
  --python-env-resource-name gruppe3-python-env \
  --driver-cores 1 --driver-memory 1g --executor-cores 1 --executor-memory 1g --num-executors 1 \
  $ENVS

# Stufe 3: der einzige echte Spark-Job. SPARK_IO_MODE=catalog schaltet die
# native Iceberg-I/O-Schicht ein (s. src/pipeline_audit_to_target.py).
cde job create --name gruppe3-stufe3-target --type spark \
  --application-file src/pipeline_audit_to_target.py \
  --mount-1-resource gruppe3-pipeline-code \
  --python-env-resource-name gruppe3-python-env \
  --driver-cores 2 --driver-memory 4g --executor-cores 2 --executor-memory 4g --num-executors 2 \
  --conf spark.kubernetes.driverEnv.SPARK_IO_MODE=catalog \
  $ENVS

cde job create --name gruppe3-stufe4-contract --type spark \
  --application-file src/contract_check.py \
  --mount-1-resource gruppe3-pipeline-code \
  --python-env-resource-name gruppe3-python-env \
  --driver-cores 1 --driver-memory 1g --executor-cores 1 --executor-memory 1g --num-executors 1 \
  $ENVS

Iceberg-Konfiguration Stufe 3: In CDE-Virtual-Clustern mit Iceberg-Support (Spark 3) sind die nötigen Katalog-Einstellungen (spark.sql.extensions, spark.sql.catalog.spark_catalog) bereits vorkonfiguriert. Falls euer VC das nicht mitbringt (erkennbar an "Table ... is not an Iceberg table" / fehlgeschlagenem INSERT OVERWRITE), dem Stufe-3-Job zusätzlich mitgeben:

--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
--conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog
--conf spark.sql.catalog.spark_catalog.type=hive

4. DAG deployen:

cde resource create --name gruppe3-dag --type files
cde resource upload --name gruppe3-dag \
  --local-path cde/pipeline_dag.py --resource-path pipeline_dag.py

cde job create --name gruppe3-pipeline-orchestrierung --type airflow \
  --dag-file pipeline_dag.py \
  --mount-1-resource gruppe3-dag

Der DAG läuft dann täglich 05:00 UTC (änderbar in cde/pipeline_dag.py, schedule_interval) und ist in der Airflow-UI des Virtual Clusters sichtbar (Monitoring, Retries, manuelles Triggern, Wiederaufsetzen ab einer gescheiterten Stufe). Er startet dank is_paused_upon_creation=False direkt aktiv — nötig, weil sich der Pause-Toggle in der CDE-Airflow-UI nicht zuverlässig bedienen ließ.

CDE-Abnahme (Checkliste nach dem Deploy)

Jeden Stufen-Job einmal einzeln manuell starten (cde job run --name gruppe3-stufe1-staging usw., Reihenfolge 0→4) und die Logs prüfen — bei unveränderten Quellen müssen die Stufen 1–3 „unveraendert - Lauf uebersprungen" bzw. „Snapshot-Fingerprints identisch" melden (der State kommt aus gruppe3_etl_state, die auch die lokalen Läufe nutzen).
Einmal Stufe 3 mit erzwungenem Build laufen lassen (--conf spark.kubernetes.driverEnv.FORCE_TARGET_BUILD=1 temporär in die Job-Config): Zeilenzahlen müssen dem letzten lokalen Rebuild entsprechen (dim_kreis 474, dim_jahr 30, dim_klimastadt 81, dim_gemeinde 10947, fact_bevoelkerung 14170, fact_bauland 4740, fact_klima 1539, fact_gemeinde_stamm 10947, fact_standortprofil_kpi 4119).
Danach den DAG einmal komplett manuell triggern — Stufe 4 muss „32 Checks OK" melden.

Hinweise:

Secrets: Die Kommandos oben legen das Workload-Passwort in die Job-Config. Wer das vermeiden will, nutzt Airflow-Connections plus einen kleinen Wrapper oder CDE-Credentials — für die Abgabe reicht die Job-Config, das Passwort ist dort nur für Projektmitglieder sichtbar.
CLI-Versionen: Die exakten Flag-Namen (--python-env-resource-name, --mount-N-resource) variieren leicht zwischen CDE-Versionen — bei Abweichungen cde job create --help konsultieren; das Setup selbst (Resources → Jobs → DAG) bleibt gleich.

Datenbank `gruppe3` auf Impala

Quellen (Datenbank default, nur lesend):

Tabelle	Inhalt	Besonderheit
`default.project_gemeinden`	10.950 Gemeinden: Name, Kreis, Fläche, Einwohner, Koordinaten	CSV-Parsing-Schäden, kein amtlicher Schlüssel
`default.project_bauland`	Baulandverkäufe je Kreis/Jahr/Merkmal (21.600 Zeilen, Langformat)	kaputte Umlaute (�), 4. Merkmal beim Import zerstört
`default.project_klimadaten`	Temperaturen je Stadt/Monat, weltweit (8,6 Mio. Zeilen)	endet 2013; Kompass-Koordinaten („53.84N")
`default.project_bevoelkerungzahlen`	Einwohner je Kreis, Breitformat (92 Spalten: 3 je Jahr 1995–2024)	kaputte Umlaute (�)

In gruppe3 erzeugt die Pipeline vier Schichten:

Schicht	Tabellen	Zweck
Staging	`gruppe3_staging_{gemeinden,bauland,klimadaten,bevoelkerungzahlen}`	unveränderte Rohkopie
Audit	`gruppe3_audit_{gemeinden,bauland,klimadaten,bevoelkerungzahlen}`	bereinigte Basis
Datenprodukt	4 × `gruppe3_dim_`, 5 × `gruppe3_fact_`	Star-Schema für Konsumenten
ETL-Metadaten	`gruppe3_etl_state` (+ kurzlebige `*_wap_incoming`-Shadow-Tabellen, existieren nur während eines Stufe-3-Builds und werden nach dem Publish gedroppt)	Incremental-Loading-Zustand + atomarer Publish (kein Konsumenten-Interface)

Alle Tabellen sind Apache-Iceberg-Tabellen (format-version 2); die Klimadaten-Tabellen sind per Iceberg-Transform TRUNCATE(4, dt) nach Jahr partitioniert. Jeder Schreibvorgang ist ein Iceberg-Snapshot — DESCRIBE HISTORY <tabelle> zeigt die Historie, SELECT … FOR SYSTEM_TIME AS OF … liest einen früheren Stand (Time Travel).

Das Datenprodukt (live verifiziert 14.07.2026, Stand nach dem Berlin/Hamburg-Fix):

Dimensionen	Fakten
`gruppe3_dim_kreis` (474)	`gruppe3_fact_bevoelkerung` (14.170, 1995–2024)
`gruppe3_dim_jahr` (30, lückenlos 1995–2024)	`gruppe3_fact_bauland` (4.740, 2015–2024)
`gruppe3_dim_gemeinde` (10.947, Brücke Kreis↔Stadt)	`gruppe3_fact_klima` (1.539, 1995–2013)
`gruppe3_dim_klimastadt` (81)	`gruppe3_fact_gemeinde_stamm` (10.947)
	`gruppe3_fact_standortprofil_kpi` (4.119, 2015–2024) — die dashboard-fertigen Cross-Table-KPIs

Gefüllte Werte in fact_standortprofil_kpi (von 4.119): wohnraumdruck_index 3.934 · baulandpreis_pro_kopf_eur 3.949 · freiflaeche_pro_einwohner_qm 3.949 · klima_angepasstes_wohnraumrisiko 3.934 · verstaedterung_index 3.450 · standortattraktivitaets_score 3.931. NULL bedeutet immer: mindestens eine Eingangsgröße fehlt (amtlich unterdrückt / kein Vorjahr / kein Klima-Match) — dokumentierte Aussage, kein Fehler.

Schema, Nutzungsregeln, gemessene Qualität und Beispiel-Queries: data/data_contract.yaml.

Datenprodukt konsumieren (Output Port)

Der Output Port sind genau die 9 gruppe3_dim_*/gruppe3_fact_*-Tabellen (nicht staging/audit) — beschrieben durch den Data Contract. Konsumenten greifen darauf über denselben Impala-Endpoint zu wie die Pipeline (LDAP + HTTP-Transport + SSL, Port 443).

SQL / Notebook: direkt per impyla/JDBC gegen gruppe3.* (Verbindung wie in src/db.py).
BI-Tools (z. B. Power BI): über den Cloudera ODBC Driver for Impala (der native Power-BI-Impala-Connector kann den HTTP-Transport/HTTP-Path des Knox-Gateways meist nicht). ODBC-DSN-Werte 1:1 aus .env: Host, Port 443, Auth = „User Name and Password" (LDAP), Transport = HTTP, HTTP Path = IMPALA_HTTP_PATH, SSL = an. Import-Modus (nicht DirectQuery, da die Zielfakten klein sind und der Cluster aus dem Ruhezustand „aufwacht"). Das Star-Schema mappt direkt: Fakten (*:1) an dim_kreis/dim_jahr, Klima an dim_klimastadt, Gemeinde-Stamm an dim_gemeinde.

Reines Power BI Online (Browser) kann private Impala-Quellen nicht direkt anbinden — dafür wird Power BI Desktop (kostenlos) zum Bauen des Modells gebraucht, danach „Veröffentlichen"; automatische Aktualisierung im Dienst erfordert zusätzlich ein On-premises Data Gateway.

Wichtigste Nutzungsregeln (vollständig im Contract):

Joins/Filter auf Kreis-Ebene nur über kreis_id — kreis_name ist mehrdeutig (36 Namen mehrfach, z. B. „Leipzig" 3× für Stadt/Landkreis/Alt-Kreis).
Alle Textwerte sind ASCII-transliteriert: mit 'Muenchen' filtern, nicht 'München'.
gemeinde_id ist ein Surrogat und wird je Pipeline-Lauf neu vergeben — nie extern persistieren.
kaufwert_je_qm_eur nicht verwenden (ab Quelle zerstört, s. Probleme.md); preis_pro_qm_eur ist der berechnete Ersatz.
standortattraktivitaets_score ist ein z-Score: nur innerhalb eines Jahres vergleichbar, negative Werte sind normal.

Datenqualität: was bereinigt wurde (Kurzfassung)

Zerstörte Umlaute (L�beck) in Bauland/Bevölkerung: das Originalzeichen ist als U+FFFD unwiederbringlich verloren → Stufe 2 entdeckt die betroffenen Kreisnamen zur Laufzeit und löst die korrekte Schreibweise automatisch über drei Referenzlisten auf; nur 8 Sonderfälle bleiben manuell gepflegt. Nach dem Lauf verifiziert: 0 verbleibende kaputte Zeichen in allen Audit-Tabellen (zuletzt 09.07.2026).
Intakte Umlaute (Gemeindenamen): echte Transliteration ä→ae/ö→oe/ü→ue, damit alle Namens-Joins dieselbe ASCII-Schreibweise verwenden.
Englische Städtenamen der Klimadaten (Munich→Muenchen …) gemappt, Kompass-Koordinaten („5.63S" → „-5,63") normalisiert, Filter auf country = 'Germany'.
Stadtstaaten Berlin & Hamburg (5-stelliger Kreisschlüssel ergänzt, P11): Die Quelle führt Berlin und Hamburg nur mit ihrem 2-stelligen Landesschlüssel (11 / 02) — einen eigenen 5-stelligen Kreisschlüssel gibt es nicht, ihre Untereinheiten sind die 8-stelligen Bezirke. Stufe 3 filtert aber bewusst auf LENGTH(id) == 5 (um die Bundesland-Aggregate und Bezirke zu entfernen) — das hat Berlin und Hamburg zuvor komplett verworfen (fehlten in dim_kreis, beiden Basis-Fakten und im KPI-Fakt; nur 14 statt 16 Bundesländer im Produkt). Fix: promote_stadtstaat_id() in pipeline_audit_to_target.py hebt 11→11000 (Berlin) und 02→02000 (Hamburg) vor dem Längen-Filter auf den kanonischen Kreisschlüssel; Bremen (04) bleibt unangetastet, da es echte 5-stellige Kreise (04011/04012) hat. Verifiziert: dim_kreis 474 (16 Bundesländer), KPI-Fakt 4.119 Zeilen mit Werten für 11000/02000, Contract-Gate 32/32 OK (14.07.2026).
Nicht behebbar, daher im Data Contract dokumentiert: amtlich unterdrückte Werte, Flächen-Rundung auf 0 (748 Zeilen), Klimadaten nur bis 2013, das beim Quellimport zerstörte Merkmal „Durchschnittlicher Kaufwert je qm". Alle Details: Probleme.md.

Prüfungswissen kompakt

Data Mesh (Theorie): kein Werkzeug, sondern ein Organisations-/ Architektur-Konzept mit 4 Prinzipien: Domain Ownership · Data as a Product · Self-Serve Data Platform · Federated Governance. Im Projekt konkret: fact_standortprofil_kpi + Data Contract = „Data as a Product"; ehrlich gemessene, dokumentierte Qualität = „Federated Governance"; die Cloudera-Plattform mit Gruppen-Datenbanken = „Self-Serve Platform".

OLTP vs. OLAP: OLTP = operativ (schreiben, normalisiert/3. NF); OLAP = analytisch (lesen/aggregieren, denormalisiert). Die Pipeline kopiert vom operativen ins analytische System; Star-Schema + Parquet sind die OLAP-Konsequenz.

Die 3 kniffligen Transformationen:

Unpivot (Bevölkerung breit→lang): explode(array(struct(…))) — 1 Spalte pro Jahr → 1 Zeile pro Jahr.
Pivot (Bauland lang→breit): DataFrame.pivot() — 1 Zeile pro Merkmal → 1 Spalte pro Merkmal.
z-Score (Standort-Score): (Wert − AVG über Jahr) / STDDEV über Jahr als Window-Funktion — macht %, €/m² und °C vergleichbar, bevor sie zu einem Score verrechnet werden. STDDEV() OVER kann Impala-SQL nicht → Spark.

Werkzeug-Rollen: Impala = die Datenbank (führt SQL aus, hält die Tabellen); impyla = Python-Draht zu Impala (DDL + Schreiben); Spark/PySpark = die Verarbeitungs-Engine für die Transformationen (DataFrame-API statt SQL-Strings — gleichwertig, besser komponierbar). Ablauf: Spark liest per JDBC → rechnet → impyla schreibt zurück (warum nicht df.write.jdbc: Probleme.md).

Batch, nicht Streaming: ein täglicher Lauf. Produktiv übernimmt das seit dem 11.07.2026 der Airflow-DAG in Cloudera Data Engineering (05:00 UTC, s. CDE-Deployment); lokal bleibt der APScheduler (00:00 Europe/Berlin) als zweite Betriebsart. Der frühere „Ausblick: Scheduler gehört auf die Plattform" ist damit umgesetzt.

Kür umgesetzt: Apache Iceberg als Open Table Format für alle Tabellen (atomare Snapshot-Commits, Time Travel, Fingerprint-Skip); echtes row-level MERGE INTO/DELETE bei den beiden Key-Audit-Tabellen (ADR-13).

Quellen der Referenzlisten

Die Referenzlisten für die Encoding-Auflösung in Stufe 2 (src/utils/):

Deutsche Kreise und Bundesländer: https://gist.github.com/leonbeckert/8332153a233a89156ecdbb3905579904
Deutsche Städtenamen: https://www.datenbörse.net/item/Liste_von_deutschen_Staedtenamen_.csv

Stand

Datenmodell (DDLs) + Begründung
Pipeline (3 Stufen, WAP, inkrementell, idempotent) + Orchestrator + Scheduler
Data Contract + technische Durchsetzung als Publish-Gate
Zweite Betriebsart CDE: Airflow-DAG deployt & aktiv (11.07.2026)
Restpunkte unten

Bereits live verifiziert (10.07.2026, gegen die echte DHBW-Datenbank): die einmalige Iceberg-Migration aller 20 Tabellen (Zeilenzahlen identisch), Stufe 1+2 mit Skip-Verhalten (unveränderte Quellen → alle Schritte übersprungen), der erzwungene Änderungspfad (DELETE+MERGE für Bauland/Bevölkerung, Full Refresh für Gemeinden — Inhalts-Prüfsummen vor/nach identisch, zweiter Lauf skippt wieder) sowie contract_check.py mit 32/32 Checks OK.

Neu (11.07.2026) — Pipeline in CDE online: Resources, die fünf Stufen-Jobs und der Airflow-DAG gruppe3_data_mesh_pipeline sind nach der Deploy-Anleitung im Virtual Cluster angelegt, der DAG ist aktiv (täglich 05:00 UTC). Deploy-Erkenntnis: Airflow legt neue DAGs standardmäßig pausiert an, und der Pause-Toggle ließ sich in der CDE-Airflow-UI nicht zuverlässig bedienen — der DAG setzt deshalb is_paused_upon_creation=False (cde/pipeline_dag.py).

Neu (14.07.2026) — Stadtstaaten-Fix verifiziert (P11): Berlin (11→11000) und Hamburg (02→02000) sind jetzt im Datenprodukt: dim_kreis 474 (16 Bundesländer), erzwungener Stufe-3-Rebuild + Contract-Gate 32/32 OK — Details s. Datenqualität.

Offene Arbeiten:

Zwei-Läufe-Abnahmetest: Stufe 3 selbst ist inzwischen bestätigt — der erzwungene Rebuild beim Berlin/Hamburg-Fix (14.07.2026) lief inkl. Contract-Gate 32/32 gegen die Iceberg-Audit-Tabellen, damit ist auch der Spark-Pfad nach dem Shadow-Swap-Umbau von overwrite_table abgedeckt. Noch offen: utils/reset_database.py → run_pipeline.py → zweiter Lauf muss überall „übersprungen" melden.
CDE-Abnahme nach Checkliste: die Stufen-Jobs einzeln laufen lassen (Skip-Meldungen prüfen), einmal erzwungener Stufe-3-Build mit Zeilenzahl-Abgleich gegen den lokalen Referenz-Rebuild, danach ein kompletter DAG-Lauf mit „32 Checks OK".
Data-Contract-Härtung (optional): weitere ausführbare quality-SQLs ergänzen, z.B. Row-Count-Minima, FK-Integrität und Composite-Key-Checks für fact_bevoelkerung und fact_klima. Der aktuelle Pflichtstand läuft und prüft bereits Schema, Required-Felder, einfache Eindeutigkeit und die vorhandenen SQL-Regeln.
dim_kreis.kreis_name mehrdeutig (Stadt/Landkreis-Zusatz wurde bei der Bereinigung entfernt, z. B. dreimal „Leipzig") → Zusatz-Spalte kreis_typ oder Originalname ergänzen; bis dahin gilt: Joins nur über kreis_id (im Data Contract dokumentiert).
Kosmetik: WindowExec-Warnung (Window ohne PARTITION BY, bei unserer Datenmenge unkritisch), log4j-ClassCastException des JDBC-Treibers (harmlos, s. docs/Probleme.md).

Warum ist alles so gebaut (und was galt früher)? → ADR.md Was ist schiefgegangen und wie wurde es gelöst? → Probleme.md

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Repository files navigation

Data Mesh & Data Engineering – Portfolioprüfung (Gruppe 3)

Architektur: der Datenfluss

Datenmodell & Begründung

Ausgangslage: vier Quelltabellen

Gewähltes Modell: Star-/Galaxy-Schema (4 Dimensionen + 5 Fakten)

Warum so? (die Begründung)

Idempotenz

Incremental Loading: wie die Pipeline Änderungen erkennt

Projektstruktur

Wie liest man den Code?

Einrichtung (einmalig)

Benutzung

Data Contract CLI

Handgeschriebener Contract vs. CLI-generiertes Gerüst

Docker

Zwei Betriebsarten: lokal (Docker) und Cloudera Data Engineering

CDE-Deployment (einmalig, CDE CLI)

CDE-Abnahme (Checkliste nach dem Deploy)

Datenbank `gruppe3` auf Impala

Datenprodukt konsumieren (Output Port)

Datenqualität: was bereinigt wurde (Kurzfassung)

Prüfungswissen kompakt

Quellen der Referenzlisten

Stand

About

Releases

Packages

Contributors

Languages

Name		Name	Last commit message	Last commit date
Latest commit History 74 Commits
airflow		airflow
data		data
docs		docs
src		src
.dockerignore		.dockerignore
.env.example		.env.example
.gitignore		.gitignore
Dockerfile		Dockerfile
README.md		README.md
docker-compose.yml		docker-compose.yml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation

Data Mesh & Data Engineering – Portfolioprüfung (Gruppe 3)

Architektur: der Datenfluss

Datenmodell & Begründung

Ausgangslage: vier Quelltabellen

Gewähltes Modell: Star-/Galaxy-Schema (4 Dimensionen + 5 Fakten)

Warum so? (die Begründung)

Idempotenz

Incremental Loading: wie die Pipeline Änderungen erkennt

Projektstruktur

Wie liest man den Code?

Einrichtung (einmalig)

Benutzung

Data Contract CLI

Handgeschriebener Contract vs. CLI-generiertes Gerüst

Docker

Zwei Betriebsarten: lokal (Docker) und Cloudera Data Engineering

CDE-Deployment (einmalig, CDE CLI)

CDE-Abnahme (Checkliste nach dem Deploy)

Datenbank gruppe3 auf Impala

Datenprodukt konsumieren (Output Port)

Datenqualität: was bereinigt wurde (Kurzfassung)

Prüfungswissen kompakt

Quellen der Referenzlisten

Stand

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Datenbank `gruppe3` auf Impala