Natuurlijk! Hier is een overzicht van belangrijke statistieken, manieren om prestaties en fouten te loggen, en voorbeelden van Python-code voor monitoring met Pandas: Belangrijke statistieken om te volgen: 1. Aantal geladen records: controleer of het verwachte aantal rijen is ingeladen. 2. Gegevenskwaliteit: aantal ontbrekende waarden per kolom. 3. Unieke waarden: bijvoorbeeld voor categorische kolommen. 4. Statistieken zoals mean, median, min, max voor numerieke kolommen. 5. Verandering in statistieken tussen verschillende ladingen (bij herhaald gebruik). 6. Fouten tijdens het laden (bijvoorbeeld verkeerde bestandsformaten). Hoe prestaties en fouten loggen: - Gebruik de Python logging-module om logs vast te leggen. - Log zowel succes- als foutmeldingen. - Voeg timestamp toe voor traceerbaarheid. - Overweeg het opslaan van statistieken in een dashboard of een bestand. Voorbeeld Python-code: ```python import pandas as pd import logging from datetime import datetime # Logging instellen logging.basicConfig( filename='monitoring.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def laad_gegevens(file_path): try: df = pd.read_csv(file_path) logging.info(f"Gegevens succesvol geladen uit {file_path}. Totaal records: {len(df)}") return df except Exception as e: logging.error(f"Fout bij het laden van {file_path}: {e}") return None def analyseer_gegevens(df): try: # Aantal ontbrekende waarden per kolom missing_values = df.isnull().sum() logging.info(f"Ontbrekende waarden per kolom:\n{missing_values}") # Unieke waarden voor categorische kolommen for col in df.select_dtypes(include='object'): unieke_waarden = df[col].nunique() logging.info(f"Kolom '{col}': {unieke waarden} unieke waarden.") # Statistieken voor numerieke kolommen numerieke_kolommen = df.select_dtypes(include='number') statistieken = numerieke_kolommen.describe() logging.info(f"Numerieke statistieken:\n{statistieken}") except Exception as e: logging.error(f"Fout tijdens analyse: {e}") # Voorbeeld gebruik bestandspad = 'gegevens.csv' gegevens = laad_gegevens(bestandspad) if gegevens is not None: analyseer_gegevens(gegevens) else: logging.warning("Geen gegevens om te analyseren.") ``` Samenvatting: - Log alle belangrijke statistieken en fouten. - Monitor de gegevenskwaliteit en prestaties regelmatig. - Pas de code aan afhankelijk van je specifieke data en vereisten. Wil je nog hulp bij het automatiseren of visualiseren van deze statistieken?

Zeker, ik help je graag met het opzetten van monitoring voor je Pandas pijplijn. Hier is een complete handleiding: ## Belangrijke statistieken om te monitoren ### 1. **Gegevenskwaliteit** - Aantal rijen en kolommen - Ontbrekende waarden per kolom - Duplicaten - Datatype distributie ### 2. **Prestatiestatistieken** - Verwerkingstijd per stap - Geheugengebruik - Bestandsgrootte ### 3. **Foutmonitoring** - Aantal verwerkte/gefaalde rijen - Validatiefouten - Uitzonderingen en fouttypes ## Implementatie met Python ```python import pandas as pd import logging import time from datetime import datetime import sys # Logging configuratie logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('pipeline_monitoring.log'), logging.StreamHandler(sys.stdout) ] ) class PipelineMonitor: def __init__(self, pipeline_name): self.pipeline_name = pipeline_name self.start_time = None self.stats = {} def start_step(self, step_name): """Start timing voor een pijplijnstap""" self.start_time = time.time() logging.info(f"START: {step_name}") return self.start_time def end_step(self, step_name, df=None, success=True): """Beëindig timing en verzamel statistieken""" if self.start_time: duration = time.time() - self.start_time step_stats = { 'duration': duration, 'timestamp': datetime.now(), 'success': success } if df is not None: step_stats.update({ 'rows': len(df), 'columns': len(df.columns), 'memory_usage': df.memory_usage(deep=True).sum(), 'null_counts': df.isnull().sum().to_dict(), 'duplicates': df.duplicated().sum() }) self.stats[step_name] = step_stats status = "VOLTOOID" if success else "GEFAALD" logging.info(f"{status}: {step_name} - Duur: {duration:.2f}s") if df is not None: logging.info(f"Statistieken: {len(df)} rijen, {len(df.columns)} kolommen") def generate_report(self): """Genereer een samenvattend rapport""" report = f"\n=== MONITORING RAPPORT: {self.pipeline_name} ===\n" total_duration = 0 for step, stats in self.stats.items(): report += f"\n{step}:\n" report += f" - Duur: {stats['duration']:.2f}s\n" report += f" - Status: {'Geslaagd' if stats['success'] else 'Gefaald'}\n" if 'rows' in stats: report += f" - Rijen: {stats['rows']}\n" report += f" - Kolommen: {stats['columns']}\n" report += f" - Geheugen: {stats['memory_usage'] / 1024**2:.2f} MB\n" report += f" - Duplicaten: {stats['duplicates']}\n" total_duration += stats['duration'] report += f"\nTotale pijplijnduur: {total_duration:.2f}s" return report # Voorbeeld pijplijn met monitoring def data_processing_pipeline(csv_file_path): monitor = PipelineMonitor("CSV Verwerkingspijplijn") try: # Stap 1: Gegevens laden monitor.start_step("CSV laden") df = pd.read_csv(csv_file_path) monitor.end_step("CSV laden", df) # Stap 2: Gegevens opschonen monitor.start_step("Gegevens opschonen") # Verwijder duplicaten df_clean = df.drop_duplicates() # Vul ontbrekende waarden in df_clean = df_clean.fillna(method='ffill') monitor.end_step("Gegevens opschonen", df_clean) # Stap 3: Gegevenstransformatie monitor.start_step("Gegevenstransformatie") # Voeg berekende kolommen toe numeric_columns = df_clean.select_dtypes(include=['number']).columns if len(numeric_columns) > 0: df_clean['total_numeric'] = df_clean[numeric_columns].sum(axis=1) monitor.end_step("Gegevenstransformatie", df_clean) # Print rapport print(monitor.generate_report()) return df_clean except Exception as e: logging.error(f"Pijplijn gefaald: {str(e)}") monitor.end_step("Pijplijn", success=False) raise # Real-time monitoring functies def log_data_quality(df, step_name): """Log gegevenskwaliteit statistieken""" quality_report = f""" === GE GEVENSKWALITEIT RAPPORT: {step_name} === Vorm: {df.shape} Ontbrekende waarden: {df.isnull().sum().sum()} Duplicaten: {df.duplicated().sum()} Geheugengebruik: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB Kolomdatatypes: {df.dtypes} """ logging.info(quality_report) def validate_dataframe(df, validation_rules): """Valideer dataframe tegen regels""" errors = [] for rule_name, rule_func in validation_rules.items(): try: if not rule_func(df): errors.append(f"Validatiefout: {rule_name}") except Exception as e: errors.append(f"Validatiefout {rule_name}: {str(e)}") if errors: for error in errors: logging.error(error) return False return True # Gebruiksvoorbeeld if __name__ == "__main__": # Definieer validatieregels validation_rules = { 'niet_leeg': lambda df: len(df) > 0, 'heeft_kolommen': lambda df: len(df.columns) > 0, 'geen_alleen_nan_rijen': lambda df: not df.isnull().all(axis=1).any() } try: # Voer pijplijn uit result_df = data_processing_pipeline('jouw_bestand.csv') # Valideer resultaat if validate_dataframe(result_df, validation_rules): logging.info("Pijplijn succesvol voltooid en gevalideerd") else: logging.warning("Pijplijn voltooid maar validatie gefaald") except Exception as e: logging.critical(f"Pijplijn uitgevallen: {str(e)}") ``` ## Aanvullende monitoring tips ### 1. **Alerting toevoegen** ```python def send_alert(message, level='INFO'): """Stuur alerts bij kritieke gebeurtenissen""" if level in ['ERROR', 'CRITICAL']: # Implementeer email/SMS/slack notificatie logging.critical(f"ALERT: {message}") ``` ### 2. **Prestatie benchmarken** ```python def benchmark_performance(df, operation_name, operation_func): """Benchmark een specifieke operatie""" start_time = time.time() result = operation_func(df) duration = time.time() - start_time logging.info(f"Benchmark {operation_name}: {duration:.2f}s voor {len(df)} rijen") return result, duration ``` ### 3. **Gegevensdrift detectie** ```python def detect_data_drift(current_df, reference_df, threshold=0.1): """Detecteer significante veranderingen in gegevensdistributie""" drift_detected = False for column in current_df.columns: if column in reference_df.columns: current_mean = current_df[column].mean() reference_mean = reference_df[column].mean() if abs(current_mean - reference_mean) / reference_mean > threshold: logging.warning(f"Data drift gedetecteerd in kolom {column}") drift_detected = True return drift_detected ``` Deze monitoring setup geeft je volledig inzicht in je pijplijnprestaties, helpt problemen vroegtijdig te detecteren en zorgt voor betrouwbare gegevensverwerking. Pas de statistieken en drempels aan op basis van je specifieke use case.