Handleiding: Data Pipelines Bouwen

1

Wat is een data pipeline & tooling kiezen

Definitie, vergelijking tools, wanneer wat gebruiken

Een data pipeline is een geautomatiseerd proces dat data verplaatst van bron naar bestemming, inclusief validatie, transformatie en foutafhandeling. Het verschil met een losse script: een pipeline is herhaalbaar, observeerbaar en herstelbaar bij fouten.

Tool	Type	Sterk in	Wanneer kiezen
Python + SQLAlchemy	Script	Flexibiliteit, eenvoud	Klein team, eenvoudige pipelines
Azure Data Factory	No-code/low-code	Azure integratie, GUI	Niet-technische teams, Azure-first
Apache Airflow	Orchestrator	Complexe afhankelijkheden, Python	Grote teams, veel pipelines
Databricks DLT	Declaratief	Streaming + batch, Delta	Lakehouse architectuur
dbt	Transformatie	SQL transformaties, tests	Transformatielaag in DWH

ℹ️ Voor de meeste organisaties: begin met Python voor eenvoudige pipelines, voeg Airflow toe zodra je meer dan 5-10 pipelines hebt die van elkaar afhangen. Azure Data Factory is een goede keuze als het team geen Python wil schrijven.

2

Eenvoudige Python pipeline

requests + pandas + SQLAlchemy, API naar database

Een Python pipeline die data ophaalt van een REST API en laadt in een database is een uitstekend startpunt en leert je de fundamenten van pipeline design.

python — requirements.txt

requests==2.31.0 pandas==2.2.0 sqlalchemy==2.0.27 psycopg2-binary==2.9.9 python-dotenv==1.0.0

python — pipeline.py

import os import logging from datetime import datetime, timezone import requests import pandas as pd from sqlalchemy import create_engine, text from dotenv import load_dotenv load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" ) log = logging.getLogger(__name__) def extract(api_url: str, api_key: str) -> list[dict]: """Haal data op van REST API met paginering.""" records = [] page = 1 headers = {"Authorization": f"Bearer {api_key}"} while True: response = requests.get( api_url, headers=headers, params={"page": page, "per_page": 100}, timeout=30 ) response.raise_for_status() # gooit HTTPError bij 4xx/5xx data = response.json() if not data.get("items"): break records.extend(data["items"]) log.info(f"Pagina {page}: {len(data['items'])} records opgehaald") page += 1 log.info(f"Extract klaar: {len(records)} records totaal") return records def transform(records: list[dict]) -> pd.DataFrame: """Valideer en normaliseer de ruwe API data.""" df = pd.DataFrame(records) # Type casting df["order_date"] = pd.to_datetime(df["order_date"], utc=True) df["amount"] = pd.to_numeric(df["amount"], errors="coerce") # Validatie invalid = df[df["amount"].isna() | (df["amount"] < 0)] if not invalid.empty: log.warning(f"{len(invalid)} ongeldige records weggegooid") df = df[df["amount"] >= 0] # Metadata toevoegen df["_loaded_at"] = datetime.now(timezone.utc) return df[["order_id", "customer_id", "order_date", "amount", "status", "_loaded_at"]] def load(df: pd.DataFrame, engine, table: str) -> None: """Laad data via upsert in de database.""" with engine.begin() as conn: # Tijdelijke staging tabel df.to_sql(f"{table}_staging", conn, if_exists="replace", index=False) # Upsert: insert + update bij conflict conn.execute(text(f""" INSERT INTO {table} (order_id, customer_id, order_date, amount, status, _loaded_at) SELECT order_id, customer_id, order_date, amount, status, _loaded_at FROM {table}_staging ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status, _loaded_at = EXCLUDED._loaded_at """)) log.info(f"Load klaar: {len(df)} rijen in '{table}'") if __name__ == "__main__": engine = create_engine(os.getenv("DATABASE_URL")) raw = extract(os.getenv("API_URL"), os.getenv("API_KEY")) clean = transform(raw) load(clean, engine, "bronze.orders")

3

Azure Data Factory pipeline

Copy Activity, Linked Services, parameters, triggers

Azure Data Factory (ADF) is Microsoft's cloud ETL service met een grafische interface. Pipelines worden opgeslagen als JSON en kunnen worden beheerd via Git.

Linked Service configureren (REST API bron)

json — linked service definitie

{ "name": "ls_rest_api_orders", "type": "RestService", "properties": { "url": "https://api.example.com", "enableServerCertificateValidation": true, "authenticationType": "Anonymous", "authHeaders": { "Authorization": { "type": "AzureKeyVaultSecret", "store": { "referenceName": "ls_keyvault" }, "secretName": "api-token" } } } }

Pipeline JSON (Copy Activity)

json — pipeline met parameters

{ "name": "pl_ingest_orders", "parameters": { "start_date": { "type": "string" }, "end_date": { "type": "string" } }, "activities": [ { "name": "copy_orders_to_adls", "type": "Copy", "inputs": [{ "referenceName": "ds_rest_orders", "parameters": { "start": "@pipeline().parameters.start_date", "end": "@pipeline().parameters.end_date" } }], "outputs": [{ "referenceName": "ds_adls_bronze_orders" }], "typeProperties": { "source": { "type": "RestSource" }, "sink": { "type": "ParquetSink", "storeSettings": { "type": "AzureBlobFSWriteSettings" } } } } ] }

✅ Sla gevoelige verbindingsgegevens (wachtwoorden, API keys) altijd op in Azure Key Vault. Gebruik Managed Identity voor ADF om zonder opgeslagen credentials te authenticeren bij Azure resources.

4

Airflow DAG schrijven

Python Operator, Sensors, TaskFlow API, scheduling

Apache Airflow is de meest gebruikte open-source orchestrationtool. DAGs (Directed Acyclic Graphs) definiëren de taken en hun afhankelijkheden.

python — airflow dag (taskflow api)

from __future__ import annotations from datetime import datetime, timedelta from airflow.decorators import dag, task from airflow.providers.http.sensors.http import HttpSensor from airflow.providers.postgres.hooks.postgres import PostgresHook default_args = { "owner": "data-team", "retries": 3, "retry_delay": timedelta(minutes=5), "email_on_failure": True, "email": ["alerts@bedrijf.nl"], } @dag( dag_id="ingest_orders_daily", schedule="0 6 * * *", # elke dag om 06:00 UTC start_date=datetime(2026, 1, 1), catchup=False, default_args=default_args, tags=["bronze", "orders"], ) def ingest_orders_daily(): # Sensor: wacht tot API beschikbaar is api_ready = HttpSensor( task_id="wait_for_api", http_conn_id="orders_api", endpoint="/health", poke_interval=60, timeout=600, ) @task() def extract(execution_date=None) -> list[dict]: import requests date_str = execution_date.strftime("%Y-%m-%d") response = requests.get( f"https://api.example.com/orders?date={date_str}", timeout=30 ) response.raise_for_status() return response.json()["items"] @task() def load_to_postgres(records: list[dict]) -> int: hook = PostgresHook(postgres_conn_id="postgres_dwh") with hook.get_conn() as conn, conn.cursor() as cur: cur.executemany( """INSERT INTO bronze.orders (order_id, customer_id, amount, status, order_date) VALUES (%(order_id)s, %(customer_id)s, %(amount)s, %(status)s, %(order_date)s) ON CONFLICT (order_id) DO NOTHING""", records ) return len(records) # Definieer afhankelijkheden raw_data = extract() row_count = load_to_postgres(raw_data) api_ready >> raw_data ingest_orders_daily()

ℹ️ Gebruik Astro CLI (Astronomer) of de officiële apache-airflow Docker Compose setup voor lokaal ontwikkelen. In productie: gebruik AWS MWAA, Google Cloud Composer of Astronomer voor managed Airflow.

5

Databricks Delta Live Tables

Declaratieve pipelines, streaming + batch, expectations

Delta Live Tables (DLT) is Databricks' declaratieve pipeline framework. Je beschrijft wat de data moet zijn, niet hoe het berekend moet worden.

python — delta live tables pipeline

import dlt from pyspark.sql import functions as F from pyspark.sql.types import StructType, StructField, StringType, DoubleType # Bronze: ruwe data inlezen @dlt.table( name="bronze_orders", comment="Ruwe orders vanuit ADLS landing zone", table_properties={"pipelines.autoOptimize.managed": "true"} ) def bronze_orders(): return ( spark.readStream .format("cloudFiles") .option("cloudFiles.format", "json") .option("cloudFiles.inferColumnTypes", "true") .load("abfss://landing@storage.dfs.core.windows.net/orders/") ) # Silver: validatie met DLT Expectations @dlt.expect_or_drop("order_id niet null", "order_id IS NOT NULL") @dlt.expect_or_drop("positief bedrag", "amount > 0") @dlt.expect("geldige status", "status IN ('open', 'closed', 'pending')") @dlt.table(name="silver_orders") def silver_orders(): return ( dlt.read_stream("bronze_orders") .select( F.col("order_id"), F.col("customer_id"), F.to_date("order_date").alias("order_date"), F.col("amount").cast("double"), F.lower("status").alias("status"), F.current_timestamp().alias("_processed_at") ) )

6

Error handling & Retry logica

Exceptions, exponential backoff, dead letter queues

Pipelines falen. Netwerk timeouts, API rate limits, schema wijzigingen — allemaal voorspelbare problemen waarvoor je van tevoren een strategie moet hebben.

python — retry met exponential backoff

import time import logging from functools import wraps log = logging.getLogger(__name__) def retry(max_attempts: int = 3, backoff_seconds: float = 2.0, exceptions: tuple = (Exception,)): """Decorator voor automatische retry met exponential backoff.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, max_attempts + 1): try: return func(*args, **kwargs) except exceptions as e: if attempt == max_attempts: log.error(f"Alle {max_attempts} pogingen mislukt: {e}") raise wait = backoff_seconds ** attempt log.warning( f"Poging {attempt} mislukt ({e}). " f"Wachten {wait:.0f}s voor volgende poging..." ) time.sleep(wait) return wrapper return decorator # Gebruik: @retry(max_attempts=3, backoff_seconds=2.0, exceptions=(requests.Timeout, requests.ConnectionError)) def fetch_page(url: str, page: int) -> dict: response = requests.get(url, params={"page": page}, timeout=30) response.raise_for_status() return response.json()

✅ Gebruik de tenacity library voor geavanceerde retry logica met jitter (willekeurige vertraging) om thundering herd problemen te voorkomen: pip install tenacity

7

Idempotentie & Checkpointing

Veilig herstarten, watermarks, state management

Een idempotente pipeline geeft hetzelfde resultaat ongeacht hoe vaak je hem uitvoert. Dit maakt veilig herstarten na fouten mogelijk.

Idempotentie patronen

Upsert (MERGE): INSERT + UPDATE bij conflict — geen duplicaten mogelijk
Truncate + Insert: Verwijder alles, herlaad. Eenvoudig maar traag voor grote tabellen
Partitie-overwrite: Vervang alleen de betreffende tijdspartitie
Watermark: Sla de laatste verwerkte waarde op, begin altijd daar

python — checkpoint/watermark patroon

import json from pathlib import Path class Checkpoint: """Sla pipeline voortgang op in een JSON bestand.""" def __init__(self, path: str): self.path = Path(path) def get(self, key: str, default=None): if not self.path.exists(): return default data = json.loads(self.path.read_text()) return data.get(key, default) def set(self, key: str, value) -> None: data = json.loads(self.path.read_text()) if self.path.exists() else {} data[key] = value self.path.write_text(json.dumps(data, default=str)) # Gebruik in pipeline: checkpoint = Checkpoint("pipeline_state.json") last_loaded = checkpoint.get("last_order_id", default=0) new_records = fetch_orders_since(last_loaded) load_to_db(new_records) # Sla voortgang op na succesvolle load if new_records: checkpoint.set("last_order_id", new_records[-1]["order_id"])

8

Monitoring & Alerting

Logging, metrics, Slack alerts, SLA tracking

Observeerbaarheid is de sleutel tot betrouwbare pipelines. Je moet altijd weten: wat is de laatste succesvolle run, hoeveel records zijn verwerkt, en wat gaat er mis?

python — slack alert bij fout

import requests import os def send_slack_alert(pipeline_name: str, error: str) -> None: """Stuur een Slack bericht bij pipeline fout.""" webhook_url = os.getenv("SLACK_WEBHOOK_URL") payload = { "blocks": [ { "type": "header", "text": {"type": "plain_text", "text": f"🚨 Pipeline fout: {pipeline_name}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": f"*Fout:* `{error}`\n*Tijd:* {datetime.now()}"} } ] } requests.post(webhook_url, json=payload, timeout=10) # In je pipeline hoofd-loop: try: run_pipeline() except Exception as e: send_slack_alert("ingest_orders", str(e)) raise

ℹ️ Gebruik OpenTelemetry voor gestandaardiseerde metrics en traces. In Airflow: gebruik de ingebouwde SLA Miss callbacks. In Azure Monitor: configureer alerts op ADF pipeline failures via Azure Monitor alerts.

9

Checklist

Verplichte controles voor productie deployment

Robuustheid

Retry logica met exponential backoff geïmplementeerd
Timeouts ingesteld op alle externe aanroepen
Foutmeldingen zijn informatief en bevatten context
Dead letter opslag voor afgewezen records

Idempotentie

Dubbel uitvoeren geeft hetzelfde resultaat
Upsert of partitie-overwrite gebruikt (geen pure INSERT)
Checkpointing/watermark geïmplementeerd
Pipeline handmatig herstarten getest na gesimuleerde fout

Observeerbaarheid

Structured logging actief (JSON format)
Verwerkt recordaantal gelogd per run
Alerting geconfigureerd bij fout
SLA monitoring actief (maximale vertraging)

Beveiliging

Secrets in Key Vault / environment variables (niet in code)
Minimale permissies voor service accounts
Verbindingen lopen via VNet of Private Endpoint

Data Pipelines Bouwen

Inhoud