Hieronder vind je Python-codevoorbeelden voor het uitvoeren van geautomatiseerde validatiecontroles op jouw DataFrame, inclusief controles op ontbrekende waarden, uitschieters, verkeerde datatypes en dubbele invoer. Ik geef ook uitleg over hoe je ze in je dataverwerkingsworkflow kunt integreren. 1. Importeren van benodigde libraries: ```python import pandas as pd import numpy as np ``` 2. Validatiecontroles: a. Ontbrekende waarden controleren: ```python # Controle op ontbrekende waarden missing_values = df.isnull().sum() print("Ontbrekende waarden per kolom:\n", missing_values) ``` Je kunt ook een alert instellen als er ontbrekende waarden zijn: ```python if missing_values.any(): print("Waarschuwing: Er zijn ontbrekende waarden.") ``` b. Uitschieters identificeren: Voor numerieke kolommen zoals 'leeftijd' en 'salaris' kun je gebruik maken van de Z-score of de IQR-methode: ```python # IQR-methode voor 'leeftijd' Q1 = df['leeftijd'].quantile(0.25) Q3 = df['leeftijd'].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR uitschieters_leeftijd = df[(df['leeftijd'] < lower_bound) | (df['leeftijd'] > upper_bound)] print("Uitschieters in 'leeftijd':\n", uitschieters_leeftijd) ``` Voor 'salaris' doe je hetzelfde: ```python Q1_salaris = df['salaris'].quantile(0.25) Q3_salaris = df['salaris'].quantile(0.75) IQR_salaris = Q3_salaris - Q1_salaris lower_bound_salaris = Q1_salaris - 1.5 * IQR_salaris upper_bound_salaris = Q3_salaris + 1.5 * IQR_salaris uitschieters_salaris = df[(df['salaris'] < lower_bound_salaris) | (df['salaris'] > upper_bound_salaris)] print("Uitschieters in 'salaris':\n", uitschieters_salaris) ``` c. Verkeerde datatypes detecteren en corrigeren: ```python # Controleer datatypes print(df.dtypes) # Bijvoorbeeld, 'datum' zou datetime moeten zijn if not pd.api.types.is_datetime64_any_dtype(df['datum']): df['datum'] = pd.to_datetime(df['datum'], errors='coerce') # Voor andere kolommen kun je vergelijkbare checks doen if not pd.api.types.is_integer_dtype(df['leeftijd']): df['leeftijd'] = pd.to_numeric(df['leeftijd'], errors='coerce').astype('Int64') ``` d. Dubbele invoer detecteren: ```python # Controle op dubbele rijen gebaseerd op alle kolommen dubbele_rijen = df[df.duplicated()] print("Dubbele rijen:\n", dubbele_rijen) # Of op basis van specifieke kolommen, bijvoorbeeld 'id' dubbele_id = df[df['id'].duplicated(keep=False)] print("Dubbele IDs:\n", dubbele_id) ``` 3. Integratie in de workflow: - Plaats deze controles aan het begin van je dataverwerkingsscript, direct nadat je de data hebt ingeladen. - Log de resultaten of stuur waarschuwingen zodat je weet welke issues er zijn. - Voor grote datasets kun je overwegen om deze controles in functies te plaatsen en geautomatiseerd te laten draaien. - Na het identificeren van problemen kun je besluiten om gegevens te corrigeren, te verwijderen of te markeren voor verdere review. voorbeeld: ```python def validate_dataframe(df): # Ontbrekende waarden if df.isnull().any().any(): print("Ontbrekende waarden gevonden.") print(df.isnull().sum()) # Uitschieters # (Hier kun je functies aanroepen zoals hierboven beschreven) # Datatype checks # (Hier kun je datatypes aanpassen indien nodig) # Dubbele rijen duplicaten = df[df.duplicated()] if not duplicaten.empty: print("Dubbele rijen gevonden.") print(duplicaten) # Gebruik validate_dataframe(df) ``` Door deze controles te automatiseren, verbeter je de datakwaliteit en voorkom je fouten in verdere analyses of modellering.

Hier is een uitgebreide implementatie voor geautomatiseerde datavalidatie in Pandas: ```python import pandas as pd import numpy as np from datetime import datetime import warnings class DataValidator: def __init__(self, df): self.df = df.copy() self.issues = {} def check_missing_values(self): """Controleer op ontbrekende waarden""" missing = self.df.isnull().sum() missing_percentage = (missing / len(self.df)) * 100 self.issues['missing_values'] = { 'count': missing[missing > 0].to_dict(), 'percentage': missing_percentage[missing_percentage > 0].to_dict() } return self def check_data_types(self): """Controleer of datatypes overeenkomen met verwachting""" expected_types = { 'id': 'int64', 'naam': 'object', 'leeftijd': 'int64', 'salaris': 'float64', 'datum': 'datetime64[ns]' } type_issues = {} for col, expected_type in expected_types.items(): if col in self.df.columns: actual_type = str(self.df[col].dtype) if actual_type != expected_type: type_issues[col] = { 'expected': expected_type, 'actual': actual_type } self.issues['data_type_issues'] = type_issues return self def check_outliers(self, method='iqr', threshold=3): """Controleer op uitschieters in numerieke kolommen""" numeric_cols = self.df.select_dtypes(include=[np.number]).columns outliers = {} for col in numeric_cols: if method == 'iqr': Q1 = self.df[col].quantile(0.25) Q3 = self.df[col].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - threshold * IQR upper_bound = Q3 + threshold * IQR outlier_count = ((self.df[col] < lower_bound) | (self.df[col] > upper_bound)).sum() if outlier_count > 0: outliers[col] = { 'count': outlier_count, 'percentage': (outlier_count / len(self.df)) * 100, 'bounds': [lower_bound, upper_bound] } self.issues['outliers'] = outliers return self def check_duplicates(self, subset=None): """Controleer op dubbele rijen""" if subset is None: subset = self.df.columns.tolist() duplicates = self.df.duplicated(subset=subset, keep=False) duplicate_count = duplicates.sum() self.issues['duplicates'] = { 'count': duplicate_count, 'percentage': (duplicate_count / len(self.df)) * 100, 'indices': self.df[duplicates].index.tolist() if duplicate_count > 0 else [] } return self def validate_age_range(self, min_age=18, max_age=100): """Specifieke validatie voor leeftijd""" if 'leeftijd' in self.df.columns: invalid_ages = ((self.df['leeftijd'] < min_age) | (self.df['leeftijd'] > max_age)).sum() self.issues['invalid_age'] = { 'count': invalid_ages, 'range': [min_age, max_age] } return self def validate_salary_range(self, min_salary=0, max_salary=1000000): """Specifieke validatie voor salaris""" if 'salaris' in self.df.columns: invalid_salaries = ((self.df['salaris'] < min_salary) | (self.df['salaris'] > max_salary)).sum() self.issues['invalid_salary'] = { 'count': invalid_salaries, 'range': [min_salary, max_salary] } return self def generate_report(self): """Genereer een gedetailleerd validatierapport""" report = { 'total_rows': len(self.df), 'total_columns': len(self.df.columns), 'validation_timestamp': datetime.now(), 'issues': self.issues } # Tel totaal aantal problemen total_issues = sum( len(issue_dict) if isinstance(issue_dict, dict) else issue_dict['count'] for issue_dict in self.issues.values() if issue_dict ) report['total_issues'] = total_issues return report # Gebruiksvoorbeeld def main(): # Voorbeeld DataFrame (vervang met jouw data) data = { 'id': [1, 2, 3, 4, 5, 6, 7, 8], 'naam': ['Jan', 'Piet', None, 'Klaas', 'Marie', 'Jan', 'Anna', 'Tom'], 'leeftijd': [25, 35, 45, 17, 28, 25, 150, 32], 'salaris': [50000.0, 75000.0, 100000.0, 30000.0, 45000.0, 50000.0, 2000000.0, 60000.0], 'datum': pd.to_datetime(['2023-01-01', '2023-02-01', '2023-03-01', None, '2023-05-01', '2023-01-01', '2023-07-01', '2023-08-01']) } df = pd.DataFrame(data) # Voer validatie uit validator = DataValidator(df) report = (validator .check_missing_values() .check_data_types() .check_outliers() .check_duplicates() .validate_age_range() .validate_salary_range() .generate_report()) # Toon rapport print("=== DATAVALIDATIE RAPPORT ===") print(f"Totaal rijen: {report['total_rows']}") print(f"Totaal problemen: {report['total_issues']}") print(f"Validatietijd: {report['validation_timestamp']}") print("\n--- Gedetailleerde problemen ---") for issue_type, details in report['issues'].items(): if details: print(f"\n{issue_type.upper()}:") print(details) # Integratie in workflow def data_processing_pipeline(df): """ Volledige dataverwerkingspipeline met geïntegreerde validatie """ # Stap 1: Data validatie validator = DataValidator(df) report = (validator .check_missing_values() .check_data_types() .check_outliers() .check_duplicates() .validate_age_range() .validate_salary_range() .generate_report()) # Stap 2: Bekijk rapport en neem beslissingen if report['total_issues'] > 0: print("Waarschuwing: Data kwaliteitsproblemen gedetecteerd!") # Log de issues of stuur alert # Voorbeeld: automatisch duplicates verwijderen if report['issues']['duplicates']['count'] > 0: df = df.drop_duplicates() print("Duplicates automatisch verwijderd") # Stap 3: Verdere verwerking # ... jouw normale dataverwerkingsstappen return df, report if __name__ == "__main__": main() ``` **Integratie in jouw workflow:** 1. **Voor verwerking**: Roep de validator aan voordat je met data-analyse begint 2. **Scheduling**: Zet de validatie in een dagelijkse/nachtelijke batch job 3. **Monitoring**: Sla rapporten op voor trendanalyse van data kwaliteit 4. **Alerting**: Stuur emails/notificaties bij kritieke problemen 5. **Documentatie**: Bewaar validatierapporten voor auditing **Aanbevolen verbeteringen:** ```python # Voeg logging toe import logging logging.basicConfig(level=logging.INFO) # Maak een functie voor automatische reparatie def auto_fix_issues(df, report): """Probeer veelvoorkomende problemen automatisch op te lossen""" df_fixed = df.copy() # Verwijder duplicates if report['issues']['duplicates']['count'] > 0: df_fixed = df_fixed.drop_duplicates() # Vervang extreme uitschieters met median for col, details in report['issues']['outliers'].items(): if details['count'] > 0: median_val = df_fixed[col].median() lower, upper = details['bounds'] df_fixed[col] = np.where( (df_fixed[col] < lower) | (df_fixed[col] > upper), median_val, df_fixed[col] ) return df_fixed ``` Deze implementatie geeft je een robuust systeem voor datakwaliteitscontrole dat eenvoudig is aan te passen en uit te breiden voor jouw specifieke use case.