Het opzetten van een batchverwerkingssysteem voor grote datasets in Python met Pandas vereist aandacht voor geheugenbeheer, prestaties en gegevensintegriteit. Hieronder geef ik een uitgebreide uitleg inclusief Python-codevoorbeelden en best practices. 1. Data opsplitsen in batches 2. Geheugengebruik beheren 3. Prestatieoptimalisatie 4. Gegevensintegriteit waarborgen --- ### 1. Data opsplitsen in batches Gebruik `pd.read_csv()` met `chunksize` om grote CSV-bestanden in kleinere brokken te lezen, of split DataFrame na het laden. **Voorbeeld:** ```python import pandas as pd # Van een grote CSV lezen in batches chunk_size = 10000 # grootte van elke batch for chunk in pd.read_csv('grote_data.csv', chunksize=chunk_size): # Verwerk elke batch hier process_chunk(chunk) ``` **Of, als je al een DataFrame hebt:** ```python def split_dataframe(df, batch_size): for start in range(0, len(df), batch_size): yield df.iloc[start:start + batch_size] # Bijvoorbeeld: batch_size = 5000 for batch in split_dataframe(je_dataframe, batch_size): process_chunk(batch) ``` --- ### 2. Geheugengebruik beheren - **Gebruik van `dtype` specificaties:** voorkom onnodig geheugengebruik door gegevens te typen. ```python dtypes = { 'Datum': 'datetime64[ns]', 'Verkoop': 'int32', 'Winst': 'int16' } df = pd.read_csv('grote_data.csv', dtype=dtypes, parse_dates=['Datum']) ``` - **Gebruik van `categorical` datatypes:** bij categorische data. ```python df['Categorie'] = df['Categorie'].astype('category') ``` - **Verwerking in streaming:** verwerk data in kleine stukken zodat je niet alles in het geheugen hoeft te laden. --- ### 3. Prestatieoptimalisatie - **Vectorisatie:** gebruik Pandas/numpy vectorisatie voor snelle bewerkingen. - **Gebruik van `numexpr`:** voor snelle uitdrukkingen. - **Parallel processing:** met bijvoorbeeld `multiprocessing` of `joblib` om batches parallel te verwerken. **Voorbeeld met `joblib`:** ```python from joblib import Parallel, delayed def process_chunk(chunk): # voorbeeldbewerking chunk['Winst_Marge'] = chunk['Winst'] / chunk['Verkoop'] return chunk results = Parallel(n_jobs=4)(delayed(process_chunk)(chunk) for chunk in split_dataframe(df, batch_size)) # Combineer resultaten indien nodig processed_df = pd.concat(results) ``` --- ### 4. Gegevensintegriteit waarborgen - **Validaties:** controleer op ontbrekende waarden, datatypes en waarden binnen verwachte bereik. ```python def validate_chunk(chunk): assert not chunk['Verkoop'].isnull().any(), "Ontbrekende Verkoopgegevens!" assert (chunk['Verkoop'] >= 0).all(), "Negatieve verkoopwaarde!" # meer validaties ``` - **Consistentie bij samenvoegen:** gebruik juiste join- of concat-methoden. - **Log alle verwerkingen:** zodat je kunt traceren wat er is gedaan. --- ### Complete voorbeeldcode ```python import pandas as pd from joblib import Parallel, delayed def split_dataframe(df, batch_size): for start in range(0, len(df), batch_size): yield df.iloc[start:start + batch_size] def validate_chunk(chunk): assert not chunk['Verkoop'].isnull().any(), "Ontbrekende Verkoopgegevens!" assert (chunk['Verkoop'] >= 0).all(), "Negatieve verkoopwaarde!" # voeg meer validaties toe indien nodig def process_chunk(chunk): validate_chunk(chunk) # voorbeeldbewerking chunk['Winst_Marge'] = chunk['Winst'] / chunk['Verkoop'] return chunk # Lading van grote dataset dtypes = { 'Datum': 'datetime64[ns]', 'Verkoop': 'int32', 'Winst': 'int16' } df = pd.read_csv('grote_data.csv', dtype=dtypes, parse_dates=['Datum']) batch_size = 5000 processed_batches = Parallel(n_jobs=4)( delayed(process_chunk)(batch) for batch in split_dataframe(df, batch_size) ) # Samenvoegen van resultaten resultaat_df = pd.concat(processed_batches) ``` --- ### Best practices samengevat: - Lees data in batches met `chunksize`. - Gebruik juiste datatypes om geheugengebruik te minimaliseren. - Verwerk data in parallel waar mogelijk. - Voer validaties uit om dataconsistentie te garanderen. - Documenteer en log je verwerkingen voor traceerbaarheid. Door deze aanpak kun je grote datasets efficiënt, betrouwbaar en performant verwerken.

Zeker! Ik help je bij het opzetten van een efficiënt batchverwerkingssysteem voor grote datasets in Pandas. Hier is een complete uitleg met codevoorbeelden en best practices. ## 1. Batchverwerking Basisimplementatie ```python import pandas as pd import numpy as np from typing import Generator, Callable class BatchProcessor: def __init__(self, chunk_size: int = 10000): self.chunk_size = chunk_size def read_in_chunks(self, file_path: str) -> Generator[pd.DataFrame, None, None]: """Lees CSV in chunks om geheugengebruik te optimaliseren""" for chunk in pd.read_csv(file_path, chunksize=self.chunk_size): yield chunk def process_chunk(self, chunk: pd.DataFrame) -> pd.DataFrame: """Verwerk een individuele chunk""" # Voorbeeld: Bereken winstmarge chunk['Winstmarge'] = (chunk['Winst'] / chunk['Verkoop']) * 100 # Voorbeeld: Voeg datumtransformatie toe chunk['Datum'] = pd.to_datetime(chunk['Datum']) chunk['Jaar'] = chunk['Datum'].dt.year chunk['Maand'] = chunk['Datum'].dt.month return chunk # Gebruiksvoorbeeld processor = BatchProcessor(chunk_size=5000) # Verwerk data in batches processed_chunks = [] for chunk in processor.read_in_chunks('grote_dataset.csv'): processed_chunk = processor.process_chunk(chunk) processed_chunks.append(processed_chunk) # Combineer resultaten (alleen als het past in geheugen) final_result = pd.concat(processed_chunks, ignore_index=True) ``` ## 2. Geheugenbeheer Optimalisaties ```python import psutil import gc class MemoryAwareBatchProcessor: def __init__(self, max_memory_percent: float = 70.0): self.max_memory_percent = max_memory_percent def get_memory_usage(self) -> float: """Huidig geheugengebruik als percentage""" return psutil.virtual_memory().percent def safe_to_process(self) -> bool: """Controleer of er voldoende geheugen beschikbaar is""" return self.get_memory_usage() < self.max_memory_percent def process_with_memory_management(self, file_path: str): """Verwerk data met geheugenbewaking""" results = [] for i, chunk in enumerate(pd.read_csv(file_path, chunksize=10000)): if not self.safe_to_process(): print(f"Geheugenlimiet bereikt bij chunk {i}. Wacht even...") gc.collect() # Forceer garbage collection # Verwerk chunk processed = self.process_chunk(chunk) results.append(processed) # Tussentijdse opschoning if i % 10 == 0: gc.collect() return pd.concat(results, ignore_index=True) # Optimaliseer datatypes voor geheugengebruik def optimize_dataframe(df: pd.DataFrame) -> pd.DataFrame: """Optimaliseer datatypes voor lager geheugengebruik""" # Converteer naar categorische types waar mogelijk for col in df.select_dtypes(include=['object']): if df[col].nunique() / len(df) < 0.5: # Als < 50% unieke waarden df[col] = df[col].astype('category') # Downcast numerieke types for col in df.select_dtypes(include=['int']): df[col] = pd.to_numeric(df[col], downcast='integer') for col in df.select_dtypes(include=['float']): df[col] = pd.to_numeric(df[col], downcast='float') return df ``` ## 3. Prestatie Optimalisaties ```python import multiprocessing as mp from concurrent.futures import ProcessPoolExecutor import dask.dataframe as dd class ParallelBatchProcessor: def __init__(self, n_workers: int = None): self.n_workers = n_workers or mp.cpu_count() - 1 def parallel_process(self, file_path: str, processing_func: Callable): """Parallelle verwerking met multiprocessing""" def process_wrapper(chunk): return processing_func(chunk) chunks = pd.read_csv(file_path, chunksize=10000) with ProcessPoolExecutor(max_workers=self.n_workers) as executor: results = list(executor.map(process_wrapper, chunks)) return pd.concat(results, ignore_index=True) # Dask voor distributed computing def process_with_dask(file_path: str): """Gebruik Dask voor zeer grote datasets""" # Dask leest automatisch in chunks ddf = dd.read_csv(file_path) # Bewerkingen worden lazy uitgevoerd ddf['Winstmarge'] = (ddf['Winst'] / ddf['Verkoop']) * 100 ddf['Datum'] = dd.to_datetime(ddf['Datum']) # Bereken resultaat (trigger computation) return ddf.compute() # Caching voor herhaalde bewerkingen from functools import lru_cache import hashlib @lru_cache(maxsize=128) def cached_processing(file_hash: str, processing_func_name: str): """Cache verwerkingsresultaten voor hergebruik""" pass ``` ## 4. Gegevensintegriteit Waarborgen ```python class DataIntegrityProcessor: def __init__(self): self.validation_errors = [] def validate_chunk(self, chunk: pd.DataFrame) -> bool: """Valideer data-integriteit per chunk""" is_valid = True # Controleer op ontbrekende waarden if chunk[['Datum', 'Verkoop', 'Winst']].isnull().any().any(): self.validation_errors.append("Ontbrekende waarden gedetecteerd") is_valid = False # Controleer op negatieve verkopen/winst if (chunk['Verkoop'] < 0).any() or (chunk['Winst'] < 0).any(): self.validation_errors.append("Negatieve waarden in Verkoop/Winst") is_valid = False # Controleer winst <= verkoop if (chunk['Winst'] > chunk['Verkoop']).any(): self.validation_errors.append("Winst groter dan Verkoop") is_valid = False return is_valid def process_with_validation(self, file_path: str): """Verwerk met uitgebreide validatie""" valid_chunks = [] for i, chunk in enumerate(pd.read_csv(file_path, chunksize=10000)): print(f"Verwerken chunk {i}...") if not self.validate_chunk(chunk): print(f"Waarschuwing: Validatiefouten in chunk {i}") # Log fouten maar ga door (of stop bij ernstige fouten) continue # Verwerk alleen valide chunks processed_chunk = self.process_chunk(chunk) valid_chunks.append(processed_chunk) if self.validation_errors: print("Validatiefouten gevonden:") for error in self.validation_errors: print(f" - {error}") return pd.concat(valid_chunks, ignore_index=True) # Transactionele verwerking met rollback mogelijkheid import tempfile import os def transactional_processing(input_file: str, output_file: str): """Transactionele verwerking met rollback bij fouten""" temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.csv') try: processor = DataIntegrityProcessor() result = processor.process_with_validation(input_file) # Schrijf naar tijdelijke file eerst result.to_csv(temp_file.name, index=False) # Verplaats naar definitieve locatie (atomische operatie) os.replace(temp_file.name, output_file) print("Verwerking succesvol voltooid") except Exception as e: # Rollback: verwijder tijdelijke file if os.path.exists(temp_file.name): os.unlink(temp_file.name) print(f"Verwerking mislukt: {e}") raise ``` ## 5. Complete Workflow Voorbeeld ```python def complete_batch_workflow(): """Complete batch verwerkingsworkflow""" # Configuratie config = { 'chunk_size': 10000, 'max_memory_usage': 75.0, 'enable_validation': True, 'output_file': 'verwerkte_data.csv' } # Initialiseer processors memory_processor = MemoryAwareBatchProcessor( max_memory_percent=config['max_memory_usage'] ) integrity_processor = DataIntegrityProcessor() try: # Stap 1: Geheugencontrole if not memory_processor.safe_to_process(): print("Onvoldoende geheugen beschikbaar") return # Stap 2: Batchverwerking met validatie if config['enable_validation']: result = integrity_processor.process_with_validation('input_data.csv') else: processor = BatchProcessor(chunk_size=config['chunk_size']) result = processor.process_with_memory_management('input_data.csv') # Stap 3: Optimaliseer geheugengebruik result = optimize_dataframe(result) # Stap 4: Transactionele opslag transactional_processing = lambda: result.to_csv(config['output_file'], index=False) transactional_processing() print("Workflow succesvol voltooid") except Exception as e: print(f"Workflow mislukt: {e}") finally: # Opruimen gc.collect() # Gebruik de workflow complete_batch_workflow() ``` ## Best Practices Samenvatting: 1. **Geheugenbeheer**: Gebruik `chunksize`, optimaliseer datatypes, monitor geheugengebruik 2. **Prestaties**: Parallelle verwerking, Dask voor zeer grote datasets, caching 3. **Integriteit**: Validatie per chunk, transactionele verwerking, error handling 4. **Monitoring**: Logging, voortgangsindicatoren, geheugengebruik tracking 5. **Schaalbaarheid**: Modulair ontwerp, configureerbare parameters, herbruikbare componenten Deze aanpak zorgt voor efficiënte verwerking van grote datasets terwijl geheugengebruik, prestaties en data-integriteit gegarandeerd blijven.