datapipeline.py

import dask.dataframe as dd
import pandas as pd
import yaml
import seaborn as sns
import matplotlib.pyplot as plt
from pathlib import Path

class DataPipeline:
    def __init__(self, config_path):
        """Initialize pipeline with config file"""
        self.config = self._load_config(config_path)
        self.dataframes = {}
        self.merged_df = None
        
    def _load_config(self, config_path):
        """Load YAML configuration file"""
        with open(config_path, 'r') as f:
            return yaml.safe_load(f)
    
    def load_data_sources(self):
        """Load all data sources defined in config"""
        sources = self.config.get('data_sources', [])
        
        for src in sources:
            name = src['name']
            path = src['path']
            file_type = src.get('type', 'csv')
            
            print(f"Loading {name} from {path}...")
            
            if file_type == 'csv':
                # Use Dask for large files
                if src.get('use_dask', False):
                    df = dd.read_csv(path)
                    df = df.compute()  # Convert to pandas
                else:
                    df = pd.read_csv(path)
            elif file_type == 'parquet':
                if src.get('use_dask', False):
                    df = dd.read_parquet(path)
                    df = df.compute()
                else:
                    df = pd.read_parquet(path)
            elif file_type == 'json':
                df = pd.read_json(path)
            else:
                raise ValueError(f"Unsupported file type: {file_type}")
            
            self.dataframes[name] = df
            print(f"  Loaded {len(df)} rows")
    
    def merge_dataframes(self):
        """Merge dataframes based on config specifications"""
        merge_specs = self.config.get('merges', [])
        
        if not merge_specs:
            print("No merge specifications found")
            return
        
        # Start with first dataframe
        result = None
        
        for spec in merge_specs:
            left_name = spec['left']
            right_name = spec['right']
            on = spec.get('on')
            how = spec.get('how', 'inner')
            left_on = spec.get('left_on')
            right_on = spec.get('right_on')
            
            left_df = self.dataframes[left_name] if result is None else result
            right_df = self.dataframes[right_name]
            
            print(f"Merging {left_name} with {right_name} on {on or (left_on, right_on)}...")
            
            if on:
                result = pd.merge(left_df, right_df, on=on, how=how)
            else:
                result = pd.merge(left_df, right_df, left_on=left_on, right_on=right_on, how=how)
            
            print(f"  Result: {len(result)} rows")
        
        self.merged_df = result
    
    def calculate_triggers(self):
        """Calculate trigger values for each trigger type"""
        trigger_config = self.config.get('triggers', {})
        df = self.merged_df if self.merged_df is not None else list(self.dataframes.values())[0]
        
        trigger_results = {}
        
        for trigger_name, trigger_spec in trigger_config.items():
            trigger_type = trigger_spec['type']
            
            print(f"\nCalculating trigger: {trigger_name} (type: {trigger_type})")
            
            if trigger_type == 'threshold':
                column = trigger_spec['column']
                threshold = trigger_spec['threshold']
                operator = trigger_spec.get('operator', '>')
                
                if operator == '>':
                    triggered = df[df[column] > threshold]
                elif operator == '<':
                    triggered = df[df[column] < threshold]
                elif operator == '>=':
                    triggered = df[df[column] >= threshold]
                elif operator == '<=':
                    triggered = df[df[column] <= threshold]
                elif operator == '==':
                    triggered = df[df[column] == threshold]
                
                trigger_results[trigger_name] = {
                    'count': len(triggered),
                    'percentage': (len(triggered) / len(df)) * 100,
                    'data': triggered
                }
                
            elif trigger_type == 'percentile':
                column = trigger_spec['column']
                percentile = trigger_spec['percentile']
                
                threshold_value = df[column].quantile(percentile / 100)
                triggered = df[df[column] >= threshold_value]
                
                trigger_results[trigger_name] = {
                    'threshold_value': threshold_value,
                    'count': len(triggered),
                    'percentage': (len(triggered) / len(df)) * 100,
                    'data': triggered
                }
                
            elif trigger_type == 'moving_average':
                column = trigger_spec['column']
                window = trigger_spec['window']
                threshold = trigger_spec['threshold']
                
                df[f'{column}_ma'] = df[column].rolling(window=window).mean()
                triggered = df[df[f'{column}_ma'] > threshold]
                
                trigger_results[trigger_name] = {
                    'count': len(triggered),
                    'percentage': (len(triggered) / len(df)) * 100,
                    'data': triggered
                }
                
            elif trigger_type == 'anomaly':
                column = trigger_spec['column']
                std_threshold = trigger_spec.get('std_threshold', 3)
                
                mean = df[column].mean()
                std = df[column].std()
                
                triggered = df[abs(df[column] - mean) > (std_threshold * std)]
                
                trigger_results[trigger_name] = {
                    'count': len(triggered),
                    'percentage': (len(triggered) / len(df)) * 100,
                    'mean': mean,
                    'std': std,
                    'data': triggered
                }
            
            print(f"  Triggered: {trigger_results[trigger_name]['count']} rows ({trigger_results[trigger_name]['percentage']:.2f}%)")
        
        return trigger_results
    
    def visualize_triggers(self, trigger_results):
        """Create visualizations using Seaborn"""
        sns.set_style("whitegrid")
        
        # Create summary plot
        trigger_names = list(trigger_results.keys())
        trigger_counts = [trigger_results[t]['count'] for t in trigger_names]
        
        fig, axes = plt.subplots(1, 2, figsize=(14, 5))
        
        # Bar plot of trigger counts
        sns.barplot(x=trigger_names, y=trigger_counts, ax=axes[0], palette='viridis')
        axes[0].set_title('Trigger Counts')
        axes[0].set_xlabel('Trigger Type')
        axes[0].set_ylabel('Count')
        axes[0].tick_params(axis='x', rotation=45)
        
        # Pie chart of trigger percentages
        trigger_percentages = [trigger_results[t]['percentage'] for t in trigger_names]
        axes[1].pie(trigger_percentages, labels=trigger_names, autopct='%1.1f%%')
        axes[1].set_title('Trigger Distribution')
        
        plt.tight_layout()
        plt.savefig('trigger_summary.png', dpi=300, bbox_inches='tight')
        print("\nVisualization saved as 'trigger_summary.png'")
        
    def run(self):
        """Execute the full pipeline"""
        print("=" * 60)
        print("DATA PIPELINE EXECUTION")
        print("=" * 60)
        
        self.load_data_sources()
        self.merge_dataframes()
        trigger_results = self.calculate_triggers()
        self.visualize_triggers(trigger_results)
        
        return trigger_results


# Example usage
if __name__ == "__main__":
    # Example config structure (create config.yaml with this structure):
    """
    data_sources:
      - name: sales
        path: data/sales.csv
        type: csv
        use_dask: true
      - name: customers
        path: data/customers.csv
        type: csv
      - name: products
        path: data/products.csv
        type: csv
      - name: regions
        path: data/regions.csv
        type: csv
      - name: metrics
        path: data/metrics.parquet
        type: parquet
      - name: events
        path: data/events.json
        type: json
    
    merges:
      - left: sales
        right: customers
        on: customer_id
        how: left
      - left: sales
        right: products
        on: product_id
        how: left
      - left: sales
        right: regions
        left_on: region_code
        right_on: code
        how: left
    
    triggers:
      high_value_sales:
        type: threshold
        column: sale_amount
        threshold: 1000
        operator: '>'
      
      top_performers:
        type: percentile
        column: performance_score
        percentile: 90
      
      revenue_trend:
        type: moving_average
        column: daily_revenue
        window: 7
        threshold: 50000
      
      outlier_detection:
        type: anomaly
        column: transaction_value
        std_threshold: 3
    """
    
    pipeline = DataPipeline('config.yaml')
    results = pipeline.run()