Natuurlijk! Hieronder vind je richtlijnen en een voorbeeld van een Python-script voor het opzetten van een gegevenspijplijn die CSV-gegevens verwerkt en naar MongoDB laadt, inclusief stappen voor extraheren, transformeren en laden (ETL). Ook geef ik tips om datakwaliteit en integriteit te waarborgen. ### 1. Te gebruiken bibliotheken - **pandas**: voor het lezen en verwerken van CSV-bestanden. - **pymongo**: voor het verbinden en werken met MongoDB. - **logging**: voor het monitoren van de pijplijn en fouten. - **datakwaliteitscontrole**: je kunt zelf functies schrijven of bestaande bibliotheken gebruiken. Installeer de benodigde bibliotheken indien nog niet gedaan: ```bash pip install pandas pymongo ``` ### 2. Voorbeeld van een ETL-script ```python import pandas as pd from pymongo import MongoClient import logging # Configureer logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Stap 1: Extractie def extract_data(file_path): try: data = pd.read_csv(file_path) logging.info(f"Data geëxtraheerd uit {file_path} met {len(data)} rijen.") return data except Exception as e: logging.error(f"Fout bij het lezen van het CSV-bestand: {e}") return pd.DataFrame() # Stap 2: Transformatie def verwijder_duplicaten(df): before = len(df) df = df.drop_duplicates() after = len(df) logging.info(f"Verwijderde duplicaten: {before - after} rijen.") return df # Extra transformatiefuncties kunnen hier worden toegevoegd def transformeer_data(df): df = verwijder_duplicaten(df) # Andere transformaties kunnen hier worden toegevoegd return df # Stap 3: Laden def load_to_mongodb(df, db_name, collection_name): try: client = MongoClient('mongodb://localhost:27017') db = client[db_name] collection = db[collection_name] # Convert DataFrame naar dictionary records data_dict = df.to_dict(orient='records') result = collection.insert_many(data_dict) logging.info(f"{len(result.inserted_ids)} documenten succesvol geladen naar MongoDB.") except Exception as e: logging.error(f"Fout bij het laden naar MongoDB: {e}") # Datakwaliteit en integriteit waarborgen def controleer_datakwaliteit(df): # Voorbeeld: controleer op ontbrekende waarden if df.isnull().values.any(): logging.warning("Ontdekte ontbrekende waarden. Overweeg imputatie of filtering.") # Voorbeeld: controleer datatypes # Voorbeeld: unieke sleutels # Voeg meer controles toe afhankelijk van vereisten # Main pipeline def main(): file_path = 'gegevens.csv' db_name = 'mijn_database' collection_name = 'mijn_collectie' # Extract df = extract_data(file_path) if df.empty: logging.error("Geen gegevens om te verwerken. Proces gestopt.") return # Controleer datakwaliteit controleer_datakwaliteit(df) # Transformeren df_t = transformeer_data(df) # Laden load_to_mongodb(df_t, db_name, collection_name) if __name__ == "__main__": main() ``` ### 3. Aanvullende richtlijnen - **Transformaties**: pas `verwijder_duplicaten()` aan en voeg andere regels toe (bijvoorbeeld kolomtransformaties, dataconversies). - **Datakwaliteit**: voeg controles toe zoals datavalidatie, controle op duplicaten vóór en na transformatie, en consistentie checks. - **Gegevensintegriteit**: gebruik transactie-achtige operaties indien nodig, en log alle stappen en fouten. - **Beveiliging**: zorg dat de verbinding met MongoDB beveiligd is, vooral in productieomgevingen. ### 4. Samenvatting - Gebruik **pandas** voor CSV-verwerking. - Definieer eenvoudige functies voor data cleaning en transformatie. - Gebruik **pymongo** voor het laden naar MongoDB. - Voeg logging toe voor monitoring. - Voer datakwaliteitcontroles uit vóór en na transformaties. Met dit raamwerk kun je een solide ETL-pijplijn opzetten die gegevens uit CSV-bestanden verwerkt en naar MongoDB laadt, terwijl je de kwaliteit en integriteit bewaakt.

Hier is een uitgebreide handleiding voor het bouwen van een gegevenspijplijn in Python: ## Aanbevolen Bibliotheken ```python # Data extractie en transformatie import pandas as pd import numpy as np # Database connectie from pymongo import MongoClient import pymongo # Data validatie import pandera as pa from datetime import datetime ``` ## Volledig Voorbeeld Script ```python import pandas as pd from pymongo import MongoClient import logging import hashlib # Setup logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class DataPipeline: def __init__(self, source_file, mongo_uri, db_name, collection_name): self.source_file = source_file self.mongo_uri = mongo_uri self.db_name = db_name self.collection_name = collection_name self.client = None self.db = None def extract(self): """Stap 1: Data extractie uit CSV bron""" try: logger.info(f"Data extractie gestart vanuit: {self.source_file}") df = pd.read_csv(self.source_file) logger.info(f"Successvol geëxtraheerd: {len(df)} rijen") return df except Exception as e: logger.error(f"Extractie gefaald: {str(e)}") raise def transform(self, df): """Stap 2: Data transformatie""" logger.info("Transformatie gestart") # Oorspronkelijk aantal rijen original_count = len(df) # 1. Verwijder duplicaten df = self.verwijder_duplicaten(df) # 2. Data cleaning en validatie df = self.clean_data(df) # 3. Bereken checksum voor integriteit df = self.add_checksum(df) logger.info(f"Transformatie voltooid: {original_count} -> {len(df)} rijen") return df def verwijder_duplicaten(self, df): """Verwijder dubbele rijen""" before = len(df) df = df.drop_duplicates() after = len(df) logger.info(f"Duplicaten verwijderd: {before - after} rijen") return df def clean_data(self, df): """Data cleaning en validatie""" # Verwijder rijen met ontbrekende essentiële waarden df = df.dropna(subset=['id']) # Pas aan op basis van je dataset # Strip whitespace van string kolommen string_columns = df.select_dtypes(include=['object']).columns for col in string_columns: df[col] = df[col].str.strip() # Converteer datums date_columns = ['date', 'created_at'] # Pas aan op basis van je dataset for col in date_columns: if col in df.columns: df[col] = pd.to_datetime(df[col], errors='coerce') return df def add_checksum(self, df): """Voeg checksum toe voor data integriteit""" df['_checksum'] = df.apply( lambda row: hashlib.md5(str(tuple(row)).encode()).hexdigest(), axis=1 ) df['_processed_at'] = datetime.now() return df def validate_data(self, df): """Data validatie""" try: # Basis validaties assert not df.empty, "DataFrame is leeg" assert '_checksum' in df.columns, "Checksum ontbreekt" # Controleer op null values in belangrijke kolommen critical_columns = ['id'] # Pas aan op basis van je dataset for col in critical_columns: if col in df.columns: null_count = df[col].isnull().sum() assert null_count == 0, f"Null values gevonden in {col}: {null_count}" logger.info("Data validatie geslaagd") return True except AssertionError as e: logger.error(f"Validatie gefaald: {str(e)}") return False def load(self, df): """Stap 3: Data laden in MongoDB""" try: # Maak verbinding met MongoDB self.client = MongoClient(self.mongo_uri) self.db = self.client[self.db_name] collection = self.db[self.collection_name] # Converteer DataFrame naar dictionary records = df.to_dict('records') # Insert data (gebruik update_one voor upsert gedrag) for record in records: # Gebruik een unieke identifier voor upsert filter_query = {'_id': record.get('id')} # Pas aan op basis van je dataset collection.update_one(filter_query, {'$set': record}, upsert=True) logger.info(f"Successvol geladen: {len(records)} records in MongoDB") except Exception as e: logger.error(f"Laden gefaald: {str(e)}") raise finally: if self.client: self.client.close() def run_pipeline(self): """Voer de volledige pijplijn uit""" try: # Extract raw_data = self.extract() # Transform transformed_data = self.transform(raw_data) # Validate if not self.validate_data(transformed_data): raise Exception("Data validatie gefaald") # Load self.load(transformed_data) logger.info("Gegevenspijplijn succesvol voltooid") except Exception as e: logger.error(f"Pijplijn gefaald: {str(e)}") raise # Gebruik van de pijplijn if __name__ == "__main__": pipeline = DataPipeline( source_file="gegevens.csv", mongo_uri="mongodb://localhost:27017", db_name="mijn_database", collection_name="mijn_collectie" ) pipeline.run_pipeline() ``` ## Aanvullende Transformatieregels ```python def aanvullende_transformaties(df): """Extra transformatie voorbeelden""" # 1. Normaliseer tekst df['naam'] = df['naam'].str.title() # 2. Categoriseer numerieke waarden df['categorie'] = pd.cut(df['waarde'], bins=5, labels=['Laag', 'Gemiddeld', 'Hoog']) # 3. Aggregeer gegevens aggregated = df.groupby('categorie').agg({ 'waarde': ['mean', 'min', 'max', 'count'] }) return df ``` ## Data Kwaliteit Waarborgen 1. **Validatie Regels**: ```python def create_validation_schema(): """Maak een data validatie schema""" schema = pa.DataFrameSchema({ "id": pa.Column(pa.Int, nullable=False), "naam": pa.Column(pa.String, nullable=False), "email": pa.Column(pa.String, pa.Check.str_matches(r'^[^@]+@[^@]+\.[^@]+$')), "leeftijd": pa.Column(pa.Int, pa.Check.in_range(0, 120)), }) return schema ``` 2. **Monitoring en Logging**: ```python # Voeg metrics toe def calculate_metrics(df): metrics = { 'total_records': len(df), 'null_percentage': df.isnull().mean() * 100, 'duplicate_count': df.duplicated().sum(), 'processing_time': datetime.now() } return metrics ``` ## Best Practices 1. **Foutafhandeling**: Gebruik try-except blokken voor elke stap 2. **Logging**: Log elke belangrijke gebeurtenis 3. **Idempotentie**: Zorg dat de pijplijn meerdere keren kan draaien 4. **Monitoring**: Implementeer health checks en alerting 5. **Testing**: Schrijf unit tests voor elke transformatie Dit script biedt een solide basis voor een robuuste gegevenspijplijn die je kunt aanpassen aan je specifieke vereisten.