Natuurlijk! Hier is een overzicht van best practices en Python-codevoorbeelden om je Pandas DataFrame te optimaliseren voor real-time analyse van verkoopgegevens: 1. Gebruik van geschikte data types 2. Data-inname optimaliseren (bijvoorbeeld via batching) 3. Indexering voor snelle query's 4. Data consolidatie en update-strategieën 5. Gebruik van efficiënte opslagformaten (bijvoorbeeld Parquet) 6. Integratie met real-time datastromen (bijvoorbeeld via streaming of database) Hieronder vind je concrete voorbeelden en uitleg: ### 1. Data types optimaliseren Gebruik de meest efficiënte datatypes om geheugen te besparen en snelheid te verhogen. ```python import pandas as pd import numpy as np # Voorbeeld van data inladen df = pd.read_csv('verkoop_data.csv', parse_dates=['tijdstempel']) # Optimaliseer datatypes df['transactiebedrag'] = pd.to_numeric(df['transactiebedrag'], downcast='float') df['transactie_id'] = pd.to_numeric(df['transactie_id'], downcast='unsigned') # Als er categorische kolommen zijn: df['product_category'] = df['product_category'].astype('category') ``` ### 2. Data-inname optimaliseren Voor real-time verwerking is batchgewijze inname of streaming effectief. ```python # Simulatie van batch-inname def batch_inname(nieuwe_data): global df nieuwe_df = pd.read_csv(nieuwe_data, parse_dates=['tijdstempel']) # Data integriteit behouden bij updates df = pd.concat([df, nieuwe_df], ignore_index=True) # Optioneel: verwijder oude data of beperk de grootte # df = df[df['tijdstempel'] > cutoff_tijd] ``` ### 3. Indexering voor snelle query's Indexeer op tijdstempel voor snelle tijdsgebaseerde query's. ```python df.set_index('tijdstempel', inplace=True) # Voor snelle filtering result = df.loc['2023-10-01':'2023-10-02'] ``` ### 4. Data bijwerken en onderhouden Gebruik `append()` of `concat()` voor updates; voor grote datasets overweeg een database zoals SQLite of een andere OLAP-database. ```python # Bijwerken van gegevens nieuwe_data = pd.read_csv('nieuwe_verkoop.csv', parse_dates=['tijdstempel']) df = pd.concat([df, nieuwe_data], ignore_index=True) df.sort_index(inplace=True) ``` ### 5. Opslagformaten Bewaar de data in kolom-geoptimaliseerde formaten zoals Parquet voor snelle opslag en lezen. ```python # Opslaan df.to_parquet('verkoop_data.parquet', compression='snappy') # Laden df = pd.read_parquet('verkoop_data.parquet') ``` ### 6. Integratie met datastromen Voor echte real-time analyses kun je overwegen om een streaming platform (zoals Kafka) te gebruiken en bijvoorbeeld PySpark of Dask voor verwerking. --- ### Samenvatting beste praktijken: - Gebruik efficiënte datatypes (`category`, downcasting numeriek). - Indexeer op tijdstempel voor snelle tijdsgebaseerde filters. - Batch-invoer en consolidatie van data voor snelheid. - Bewaar data in kolomgeoptimaliseerde opslagformaten. - Overweeg een database of streaming platform voor continue data-inname en -verwerking. - Zorg voor data-integriteit bij updates door concatenatie en sortering. Wil je een voorbeeld van een complete pijplijn inclusief data-inname, opslag en query's?

Hier is een uitgebreide handleiding voor het optimaliseren van je Pandas DataFrame voor real-time analyse, met codevoorbeelden en best practices: ## 1. Data Types Optimalisatie ```python import pandas as pd import numpy as np # Voorbeeld DataFrame df = pd.DataFrame({ 'timestamp': pd.date_range('2024-01-01', periods=100000, freq='1min'), 'transactie_id': range(100000), 'bedrag': np.random.uniform(1, 1000, 100000), 'klant_id': np.random.randint(1, 5000, 100000), 'categorie': np.random.choice(['A', 'B', 'C', 'D'], 100000) }) # Optimaliseer data types def optimaliseer_data_types(df): # Converteer naar efficiëntere types df['timestamp'] = pd.to_datetime(df['timestamp']) df['klant_id'] = df['klant_id'].astype('int32') df['bedrag'] = df['bedrag'].astype('float32') df['categorie'] = df['categorie'].astype('category') return df df = optimaliseer_data_types(df) print(f"Geheugengebruik na optimalisatie: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB") ``` ## 2. Real-time Data Inname Pipeline ```python from collections import deque import threading from queue import Queue class RealTimeDataProcessor: def __init__(self, buffer_size=1000): self.buffer_size = buffer_size self.data_buffer = deque(maxlen=buffer_size) self.lock = threading.Lock() self.df = pd.DataFrame() def voeg_transactie_toe(self, transactie_data): """Voeg real-time transactie toe met thread safety""" with self.lock: self.data_buffer.append(transactie_data) # Batch verwerking bij buffer vol if len(self.data_buffer) >= self.buffer_size: self._verwerk_buffer() def _verwerk_buffer(self): """Verwerk buffer naar hoofd DataFrame""" if self.data_buffer: nieuwe_data = pd.DataFrame(list(self.data_buffer)) nieuwe_data = optimaliseer_data_types(nieuwe_data) # Voeg toe aan bestaande DataFrame self.df = pd.concat([self.df, nieuwe_data], ignore_index=True) self.data_buffer.clear() # Onderhoud optimalisaties self._onderhoud_dataframe() def _onderhoud_dataframe(self): """Voer periodiek onderhoud uit""" # Sorteer op timestamp voor snellere queries self.df = self.df.sort_values('timestamp').reset_index(drop=True) # Creëer index voor veelgebruikte queries if 'timestamp' in self.df.columns: self.df = self.df.set_index('timestamp', drop=False) ``` ## 3. Geoptimaliseerde Query Methodes ```python class GeoptimaliseerdeAnalyse: def __init__(self, dataframe): self.df = dataframe.copy() self._setup_indexes() def _setup_indexes(self): """Creëer geoptimaliseerde indexes""" # Multi-index voor veelvoorkomende query patronen if 'timestamp' in self.df.columns and 'categorie' in self.df.columns: self.df = self.df.set_index(['timestamp', 'categorie'], drop=False) def query_tijdsperiode(self, start_tijd, eind_tijd): """Snel queryen op tijdsperiode""" mask = (self.df['timestamp'] >= start_tijd) & (self.df['timestamp'] <= eind_tijd) return self.df.loc[mask] def real_time_aggregaties(self, venster_grootte='5T'): """Real-time aggregaties met rolling windows""" return self.df.set_index('timestamp').groupby('categorie')['bedrag'].rolling(venster_grootte).agg({ 'totaal_bedrag': 'sum', 'aantal_transacties': 'count', 'gemiddelde_bedrag': 'mean' }).reset_index() def efficiënte_groupby(self, groep_kolommen, agg_dict): """Geoptimaliseerde groupby operaties""" return self.df.groupby(groep_kolomens).agg(agg_dict) ``` ## 4. Data Integriteit en Validatie ```python class DataValidator: @staticmethod def valideer_transactie(transactie): """Valideer individuele transactie""" validatie_regels = { 'bedrag': lambda x: x > 0 and x < 1000000, # Realistische bedragen 'timestamp': lambda x: pd.to_datetime(x) <= pd.Timestamp.now(), 'klant_id': lambda x: x > 0 } for veld, regel in validatie_regels.items(): if veld in transactie and not regel(transactie[veld]): raise ValueError(f"Ongeldige waarde voor {veld}: {transactie[veld]}") return True @staticmethod def data_quality_checks(df): """Voer data quality checks uit""" checks = { 'null_values': df.isnull().sum().sum() == 0, 'negative_amounts': (df['bedrag'] < 0).sum() == 0, 'future_dates': (df['timestamp'] > pd.Timestamp.now()).sum() == 0 } return all(checks.values()) ``` ## 5. Complete Real-time Pipeline ```python class RealTimeVerkoopAnalyse: def __init__(self): self.processor = RealTimeDataProcessor() self.validator = DataValidator() self.analyse = None def verwerk_real_time_transactie(self, transactie_data): """Complete verwerkingspijplijn voor real-time transacties""" try: # Validatie self.validator.valideer_transactie(transactie_data) # Toevoegen aan processor self.processor.voeg_transactie_toe(transactie_data) # Update analyses self._update_analyses() return True except Exception as e: print(f"Fout bij verwerken transactie: {e}") return False def _update_analyses(self): """Update real-time analyses""" if not self.processor.df.empty: self.analyse = GeoptimaliseerdeAnalyse(self.processor.df) def get_real_time_insights(self, venster='10T'): """Genereer real-time inzichten""" if self.analyse is None: return {} recente_data = self.analyse.query_tijdsperiode( pd.Timestamp.now() - pd.Timedelta(venster), pd.Timestamp.now() ) insights = { 'totaal_omzet': recente_data['bedrag'].sum(), 'aantal_transacties': len(recente_data), 'gemiddelde_transactie': recente_data['bedrag'].mean(), 'top_categorieën': recente_data.groupby('categorie')['bedrag'].sum().nlargest(3) } return insights # Gebruik voorbeeld pipeline = RealTimeVerkoopAnalyse() # Simuleer real-time data for i in range(100): transactie = { 'timestamp': pd.Timestamp.now(), 'transactie_id': i, 'bedrag': np.random.uniform(10, 500), 'klant_id': np.random.randint(1, 1000), 'categorie': np.random.choice(['A', 'B', 'C']) } pipeline.verwerk_real_time_transactie(transactie) # Real-time inzichten insights = pipeline.get_real_time_insights() print("Real-time Insights:", insights) ``` ## Best Practices Samenvatting: 1. **Data Types**: Gebruik `category`, `int32`, `float32` waar mogelijk 2. **Indexering**: Creëer indexes voor veelgebruikte query patronen 3. **Batch Verwerking**: Verwerk data in batches voor betere performance 4. **Thread Safety**: Gebruik locks voor concurrente toegang 5. **Validatie**: Valideer data bij inname voor integriteit 6. **Periodiek Onderhoud**: Sorteer en optimaliseer regelmatig 7. **Memory Management**: Gebruik `deque` voor buffering 8. **Query Optimalisatie**: Gebruik vectorized operations en efficiënte groupby Deze aanpak zorgt voor een schaalbare, efficiënte real-time analyse pipeline die data-integriteit behoudt tijdens updates.