Python for Data Engineers

Introductie

Python is in vrijwel elk voorgaand hoofdstuk al zijdelings gebruikt — extractiescripts in Introduction to Data Engineering, Snowpark in Snowflake, MLflow in Databricks. Dit hoofdstuk behandelt Python zelf, specifiek zoals het in data engineering wordt gebruikt: niet als taal om zware dataverwerking mee te doen (dat is het domein van SQL binnen het warehouse, of PySpark voor gedistribueerde verwerking — het onderwerp van het volgende hoofdstuk), maar als verbindende taal: API's aanroepen, bestanden verwerken, orchestratielogica schrijven, en de lijm tussen systemen die geen gedeelde taal spreken.

Dit is geen algemene Python-cursus — als je al basiskennis hebt van functies, loops en dictionaries, behandelt dit hoofdstuk specifiek de patronen die in productiepipelines steeds terugkomen en die generieke Python-tutorials zelden behandelen.

Waarom Python de Standaard Werd

Python's dominantie in data engineering is geen toeval: de leesbare syntax verlaagt de drempel voor teams die niet primair uit software engineers bestaan, het ecosysteem aan libraries (requests, pandas, SQLAlchemy, elke cloud-SDK) dekt vrijwel elke integratiebehoefte zonder zelf iets te hoeven bouwen, en het is de gemeenschappelijke taal tussen data engineering, data science en ML engineering — relevant voor de vervagende grens tussen die rollen die in Introduction to Data Engineering werd besproken. Geen enkele van deze redenen gaat over ruwe uitvoeringssnelheid — Python is trager dan Java of Scala — maar in data engineering is ontwikkelsnelheid en integratiegemak vrijwel altijd belangrijker dan de laatste milliseconden performance, omdat de zware rekenkracht toch al gedelegeerd wordt aan het warehouse of Spark-cluster.

Virtual Environments en Dependency Management

Een pipeline die op je eigen laptop werkt maar in productie faalt omdat een library-versie verschilt, is een van de meest voorkomende en makkelijkst te voorkomen operationele problemen. Een virtual environment isoleert de Python-packages van één project van die van andere projecten en van het systeem-Python:

python -m venv .venv
source .venv/bin/activate       # Windows: .venv\Scripts\activate
pip install -r requirements.txt

requirements.txt legt exacte versies vast, zodat "het werkt bij mij" wordt vervangen door "het werkt met precies deze set versies, overal":

requests==2.32.3
pandas==2.2.2
pyarrow==17.0.0

Modernere projecten gebruiken steeds vaker pyproject.toml met tools als Poetry of uv, die naast dependency-versies ook een reproduceerbare lock-file bijhouden — functioneel hetzelfde doel (reproduceerbaarheid), met betere afhankelijkheidsresolutie dan een handmatig onderhouden requirements.txt. Voor CI/CD (zie CI/CD for Data Engineering) is dit niet optioneel: zonder vastgelegde versies kan een pipeline die vorige week werkte, vandaag falen puur omdat een dependency een nieuwe, incompatibele versie uitbracht.

Robuuste API-extractie: Paginering en Retries

Een realistisch extractiescript tegen een externe API moet met twee dingen omgaan die tutorials vaak overslaan: paginering (de meeste API's geven data in brokken van bijvoorbeeld 100 records terug, niet alles in één keer) en tijdelijke fouten (netwerkhaperingen, rate limits, korte uitval) — exact de "retries zijn de norm, niet de uitzondering"-les uit Introduction to Data Engineering, hier concreet in code.

import time
import requests
from requests.exceptions import RequestException

def fetch_all_orders(api_url: str, api_key: str, max_retries: int = 3) -> list[dict]:
    """Haalt alle orders op via paginering, met exponential backoff bij fouten."""
    all_orders = []
    page = 1

    while True:
        for attempt in range(max_retries):
            try:
                response = requests.get(
                    api_url,
                    headers={"Authorization": f"Bearer {api_key}"},
                    params={"page": page, "page_size": 100},
                    timeout=30,
                )
                response.raise_for_status()
                break
            except RequestException as e:
                if attempt == max_retries - 1:
                    raise
                wait = 2 ** attempt   # exponential backoff: 1s, 2s, 4s
                time.sleep(wait)

        data = response.json()
        if not data["results"]:
            break

        all_orders.extend(data["results"])
        page += 1

    return all_orders

Merk op: de for attempt in range(max_retries)-lus vangt alleen tijdelijke fouten op en herprobeert; als alle pogingen falen, wordt de fout alsnog doorgegeven (raise) in plaats van stilzwijgend te falen — een pipeline die een aanhoudende fout negeert, is gevaarlijker dan een pipeline die crasht en een alert triggert.

Bestandsformaten in de Praktijk

Data engineers werken voortdurend met bestandsformaten, en de keuze heeft directe gevolgen voor de rest van de pipeline. CSV is universeel leesbaar maar heeft geen ingebouwd schema — elke kolom is tekst totdat iemand expliciet cast, precies het "cast nooit te vroeg"-probleem uit Data Warehousing Fundamentals. JSON behoudt structuur en geneste data, maar is rij-georiënteerd en dus inefficiënt voor analytische verwerking op schaal. Parquet is het formaat waar de meeste bronze-lagen in moderne pipelines op landen: kolomgeoriënteerd (dezelfde reden als in Data Warehousing Fundamentals besproken), zelfbeschrijvend qua schema, en sterk gecomprimeerd.

import pandas as pd

# CSV inlezen, EXPLICIET met dtype-controle -- niet blind vertrouwen op auto-detectie
df = pd.read_csv("orders.csv", dtype={"order_id": str, "customer_id": str})

# Wegschrijven als Parquet: kolomgeoriënteerd, gecomprimeerd, schema-bewust
df.to_parquet("orders.parquet", engine="pyarrow", compression="snappy")

De vuistregel: gebruik CSV/JSON voor uitwisseling met externe systemen die het vereisen, maar converteer zo vroeg mogelijk in de pipeline naar Parquet zodra data intern verder stroomt — elke stap die daarna nog CSV gebruikt, betaalt onnodig de kosten van een rij-georiënteerd, ongecomprimeerd formaat.

Type Hints: Belangrijker in Pipelines dan Elders

Python is dynamisch getypeerd, maar type hints (sinds Python 3.5) laten je functiesignaturen expliciet documenteren zonder de dynamische aard van de taal te verliezen:

def calculate_revenue(orders: list[dict[str, float]], tax_rate: float = 0.21) -> float:
    return sum(order["amount"] for order in orders) * (1 + tax_rate)

In een data pipeline is dit specifiek waardevoller dan in bijvoorbeeld een webapplicatie: een pipeline verwerkt vaak data van buiten je controle (een API die stilletjes een veldtype wijzigt, een CSV met een onverwachte lege waarde), en een tool als mypy die type hints statisch controleert, vangt een categorie fouten (een functie die een string krijgt waar een float werd verwacht) vóórdat de pipeline draait, in plaats van pas bij een crash — of erger, bij stilzwijgend foute berekeningen — middenin een nachtelijke run.

Foutafhandelingspatronen voor Pipelines

Een pipeline-specifiek patroon dat generieke Python-tutorials zelden behandelen: onderscheid maken tussen fouten die het waard zijn om te herproberen en fouten die dat niet zijn, via custom exceptions:

class RetryableError(Exception):
    """Tijdelijke fout: netwerktimeout, rate limit -- opnieuw proberen heeft zin."""
    pass

class FatalError(Exception):
    """Structurele fout: ongeldige credentials, corrupt schema -- opnieuw proberen helpt niet."""
    pass

def process_batch(batch: list[dict]) -> None:
    try:
        validate_schema(batch)      # kan FatalError opwerpen
        load_to_warehouse(batch)    # kan RetryableError opwerpen
    except FatalError:
        raise                        # nooit automatisch herproberen
    except RetryableError:
        retry_with_backoff(process_batch, batch)

Dit voorkomt de veelgemaakte fout van een generieke except Exception die elke fout hetzelfde behandelt — een verkeerd wachtwoord verdient een onmiddellijke, duidelijke melding, geen drie herhaalde pogingen die toch nooit gaan slagen. Combineer dit met context managers (with-statements) voor resources die altijd netjes afgesloten moeten worden, ook bij een fout halverwege:

from contextlib import contextmanager

@contextmanager
def warehouse_connection(conn_string: str):
    conn = connect(conn_string)
    try:
        yield conn
    finally:
        conn.close()   # gegarandeerd uitgevoerd, ook als er een exception optreedt

with warehouse_connection(CONN_STRING) as conn:
    load_data(conn, batch)

Logging in Plaats van print()

Een print()-statement verdwijnt zodra een pipeline in productie draait zonder interactieve terminal; het Python logging-module legt berichten gestructureerd vast, met niveaus (INFO, WARNING, ERROR) die filtering en routing naar monitoringtools mogelijk maken — de basis waarop de observability-principes uit Monitoring & Observability daadwerkelijk gebouwd worden:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("orders_pipeline")

logger.info(f"Extractie gestart: {len(orders)} orders opgehaald")
logger.warning(f"{skipped_count} rijen overgeslagen wegens ontbrekende customer_id")
logger.error(f"Laden mislukt na {max_retries} pogingen: {e}")

Het verschil met print() is niet alleen esthetisch: logging laat je output routeren naar een bestand, een centrale logaggregator, of een monitoringtool, met niveau-gebaseerde filtering (WARNING en hoger naar een alertingkanaal, INFO alleen naar een archief) — precies het mechanisme dat het mogelijk maakt om te weten dat een pipeline een probleem had, zonder dat iemand continu meekijkt.

Data Valideren met Dataclasses en Pydantic

Type hints alleen worden niet afgedwongen tijdens runtime — een functie die een float verwacht, crasht niet automatisch als je er een string aan doorgeeft, tenzij je dat expliciet controleert. Voor data die van buiten je controle komt (een API-response, een CSV-rij) is runtime-validatie wél nodig, en Pydantic is hiervoor de gangbare library geworden: je definieert een model met getypeerde velden, en Pydantic valideert en converteert binnenkomende data automatisch, met een duidelijke foutmelding zodra iets niet klopt:

from pydantic import BaseModel, field_validator

class Order(BaseModel):
    order_id: str
    customer_id: str
    amount: float
    status: str

    @field_validator("amount")
    @classmethod
    def amount_must_be_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("amount moet positief zijn")
        return v

# Valideert en cast automatisch; werpt een duidelijke ValidationError bij een probleem
raw_record = {"order_id": "123", "customer_id": "456", "amount": "99.50", "status": "paid"}
order = Order.model_validate(raw_record)   # amount wordt automatisch naar float gecast

Dit is de Python-equivalent van de data contracts uit Modern Data Platform Architecture en de generic/singular tests uit dbt Fundamentals, maar dan toegepast op het moment dat data een Python-pipeline binnenkomt, vóórdat het ooit een warehouse bereikt — een vroege, expliciete validatielaag in plaats van erop vertrouwen dat foute data zich pas stroomafwaarts in een dbt-test manifesteert. Standaardbibliotheek-dataclasses bieden een lichtere, valideringsvrije variant voor situaties waar je alleen structuur wilt vastleggen zonder de overhead van runtime-validatie.

Best Practices

• Gebruik altijd een virtual environment met vastgelegde versies, ook voor kleine scripts — de kosten zijn laag, de voorkomen problemen zijn kostbaar.
• Converteer zo vroeg mogelijk naar Parquet zodra data intern verder stroomt, en bewaar CSV/JSON alleen waar externe systemen dat vereisen.
• Voeg type hints toe aan elke functie die pipeline-data verwerkt, en draai mypy in CI zodat typefouten vóór productie worden gevangen.
• Onderscheid retryable en fatale fouten expliciet met custom exceptions, in plaats van elke fout hetzelfde te behandelen.
• Gebruik logging, nooit print(), in elk stuk code dat ooit ongezien in productie zal draaien.
• Valideer externe data expliciet met Pydantic (of vergelijkbaar) zodra die een pipeline binnenkomt, in plaats van te vertrouwen dat foute data pas later, dieper in het platform, wordt opgevangen.

Veelgemaakte Fouten

• Geen requirements.txt of vergelijkbaar vastleggen, waardoor een pipeline die vandaag werkt volgende week kan breken door een stille dependency-update.
• CSV blijven gebruiken tot diep in de pipeline in plaats van vroeg naar Parquet te converteren, wat onnodige I/O- en opslagkosten oplevert.
• Een brede except Exception: pass die elke fout — tijdelijk of fataal — stilzwijgend negeert, wat een pipeline "laat slagen" terwijl hij feitelijk data verliest.
• API-paginering vergeten, waardoor een extractiescript alleen de eerste pagina ophaalt en niemand merkt dat 95% van de data ontbreekt totdat een analist vraagt waarom de cijfers zo laag zijn.
• print()-statements in productiecode laten staan in plaats van naar logging te migreren, waardoor er geen bruikbare, doorzoekbare historie van wat een pipeline deed overblijft.

Performance Tips

• Gebruik pandas of pyarrow voor bestandsverwerking, niet handmatige regel-voor-regel string-parsing — de geoptimaliseerde, in C geïmplementeerde routines zijn een veelvoud sneller dan pure Python-loops.
• Batch API-aanroepen waar de bron dat toelaat (grotere page_size) in plaats van veel kleine, sequentiële requests — minder round-trips betekent minder overhead.
• Vermijd het volledig inladen van zeer grote bestanden in geheugen — gebruik chunked reading (pd.read_csv(..., chunksize=...)) of stap over naar Spark zodra volumes de grens van comfortabel in-memory verwerken overschrijden, het onderwerp van het volgende hoofdstuk.
• Cache dure, herhaalde berekeningen of lookups binnen een run expliciet, in plaats van dezelfde bewerking meerdere keren uit te voeren.

Interviewvragen

"Waarom is Python de dominante taal in data engineering, ondanks dat het trager is dan Java of Scala?" Let op: leesbaarheid, ecosysteem, en het feit dat zware rekenkracht toch wordt gedelegeerd aan het warehouse of Spark — ontwikkelsnelheid weegt zwaarder dan ruwe executiesnelheid.

"Hoe zou je een extractiescript schrijven dat robuust omgaat met een gepagineerde API en tijdelijke netwerkfouten?" Let op: een paginering-lus die doorgaat tot een lege pagina, gecombineerd met retries met exponential backoff die alleen tijdelijke fouten opvangen, niet fatale.

"Waarom zou je Parquet verkiezen boven CSV in een data pipeline?" Let op: kolomgeoriënteerd (efficiënter voor analytische verwerking), schema-bewust, en sterk gecomprimeerd — dezelfde voordelen als columnar storage in een warehouse, hier toegepast op bestandsniveau.

"Wat is het verschil tussen een retryable en een fatale fout, en waarom moet je pipeline-code dat onderscheid maken?" Let op: tijdelijke fouten (netwerktimeout) verdienen automatische herpoging; structurele fouten (verkeerde credentials, corrupt schema) verdienen dat niet — alles hetzelfde behandelen leidt tot zinloze retries of gemiste directe alerts.

"Waarom is logging beter dan print() in productiepipelines?" Let op: logging biedt niveaus, routing naar bestanden/monitoringtools, en filtering — print()-output verdwijnt zodra een pipeline zonder interactieve terminal draait.

"Type hints worden niet afgedwongen tijdens runtime — hoe valideer je dan toch data die van buiten komt?" Let op: het besef dat statische type hints en runtime-validatie twee verschillende dingen zijn, en dat een library als Pydantic expliciete, runtime-afgedwongen validatie met duidelijke foutmeldingen toevoegt voor data van onbetrouwbare bronnen.

"Wat is het verschil tussen een dataclass en een Pydantic-model?" Let op: dataclasses leggen alleen structuur vast zonder validatie; Pydantic-modellen valideren en casten actief tijdens runtime, met een duidelijke ValidationError bij een probleem.

Relevante Documentatie

• Python: venv — Virtual Environments — officiële documentatie van Python's ingebouwde virtual-environment-tool.
• Requests: HTTP for Humans — de meest gebruikte Python-library voor API-aanroepen.
• Apache Arrow: Reading and Writing Parquet — pyarrow's Parquet-ondersteuning in detail.
• Python: logging — Logging Facility — officiële documentatie van het logging-framework.
• mypy Documentation — statische type-checking voor Python.
• Pydantic Documentation — runtime-validatie en -parsing van data-modellen.

Samenvatting

Python's rol in data engineering is die van verbindende taal: API's aanroepen, bestanden verwerken, orchestratielogica schrijven — niet de zware dataverwerking zelf, die aan het warehouse of aan Spark wordt gedelegeerd. Virtual environments met vastgelegde dependency-versies voorkomen het klassieke "werkt bij mij niet in productie"-probleem; robuuste API-extractie vereist expliciete paginering en retries met onderscheid tussen tijdelijke en fatale fouten; Parquet is het standaardformaat zodra data intern stroomt. Type hints, Pydantic-validatie en logging zijn geen esthetische keuzes maar directe hulpmiddelen om fouten vroeg te vangen — bij voorkeur al bij binnenkomst van externe data, niet pas wanneer een dbt-test diep in silver iets onverwachts opmerkt — en pipelines observeerbaar te maken. Met deze Python-fundamenten op zak is de volgende stap PySpark: dezelfde taal, maar dan ingezet voor gedistribueerde verwerking op een schaal die één machine niet meer aankan.