ML Data Pipelines & Orchestration

# Basic Pipeline with Python
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
import joblib
from datetime import datetime

class BasicMLPipeline:
    def __init__(self):
        self.scaler = StandardScaler()
        self.model = RandomForestClassifier(n_estimators=100)
        self.is_trained = False
    
    def load_data(self, data_path: str) -> pd.DataFrame:
        """Load data from various sources"""
        if data_path.endswith('.csv'):
            return pd.read_csv(data_path)
        elif data_path.endswith('.parquet'):
            return pd.read_parquet(data_path)
        else:
            raise ValueError("Unsupported file format")
    
    def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """Basic data cleaning"""
        # Remove duplicates
        df = df.drop_duplicates()
        
        # Handle missing values
        numeric_columns = df.select_dtypes(include=['number']).columns
        df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].median())
        
        # Remove outliers (simple IQR method)
        for col in numeric_columns:
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            df = df[(df[col] >= lower_bound) & (df[col] <= upper_bound)]
        
        return df
    
    def feature_engineering(self, df: pd.DataFrame) -> pd.DataFrame:
        """Basic feature engineering"""
        # Create interaction features
        if 'feature1' in df.columns and 'feature2' in df.columns:
            df['feature1_x_feature2'] = df['feature1'] * df['feature2']
        
        # Create time-based features if timestamp exists
        if 'timestamp' in df.columns:
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            df['hour'] = df['timestamp'].dt.hour
            df['day_of_week'] = df['timestamp'].dt.dayofweek
            df['is_weekend'] = df['day_of_week'].isin([5, 6])
        
        return df
    
    def train_pipeline(self, data_path: str, target_column: str):
        """Full training pipeline"""
        print(f"Starting training pipeline at {datetime.now()}")
        
        # 1. Load data
        df = self.load_data(data_path)
        print(f"Loaded {len(df)} records")
        
        # 2. Clean data
        df = self.clean_data(df)
        print(f"After cleaning: {len(df)} records")
        
        # 3. Feature engineering
        df = self.feature_engineering(df)
        
        # 4. Prepare features and target
        X = df.drop(columns=[target_column])
        y = df[target_column]
        
        # 5. Split data
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        
        # 6. Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)
        
        # 7. Train model
        self.model.fit(X_train_scaled, y_train)
        self.is_trained = True
        
        # 8. Evaluate
        train_score = self.model.score(X_train_scaled, y_train)
        test_score = self.model.score(X_test_scaled, y_test)
        
        print(f"Training completed:")
        print(f"Train accuracy: {train_score:.3f}")
        print(f"Test accuracy: {test_score:.3f}")
        
        # 9. Save model
        self.save_model()
        
        return {
            'train_score': train_score,
            'test_score': test_score,
            'model_path': 'model.joblib'
        }
    
    def save_model(self, path: str = 'model.joblib'):
        """Save trained model and scaler"""
        if not self.is_trained:
            raise ValueError("Model must be trained before saving")
        
        model_data = {
            'model': self.model,
            'scaler': self.scaler,
            'trained_at': datetime.now().isoformat()
        }
        
        joblib.dump(model_data, path)
        print(f"Model saved to {path}")