Real-time Analytics &ML Pipelines

// Feature Store Architecture with Redis + PostgreSQL
@Service
public class FeatureStoreService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private FeatureRepository featureRepository;
    
    // Online serving (real-time)
    public Map<String, Object> getFeaturesOnline(String entityId, List<String> featureNames) {
        Map<String, Object> features = new HashMap<>();
        
        for (String featureName : featureNames) {
            String key = String.format("feature:%s:%s", entityId, featureName);
            Object value = redisTemplate.opsForValue().get(key);
            
            if (value != null) {
                features.put(featureName, value);
            } else {
                // Fallback to database for missing features
                Feature feature = featureRepository.findByEntityIdAndName(entityId, featureName);
                if (feature != null) {
                    features.put(featureName, feature.getValue());
                    // Cache for future requests
                    redisTemplate.opsForValue().set(key, feature.getValue(), Duration.ofHours(1));
                }
            }
        }
        
        return features;
    }
    
    // Offline serving (batch)
    public Dataset<Row> getFeaturesOffline(SparkSession spark, String entityType, List<String> featureNames) {
        // Read from feature tables for batch processing
        String featureTable = String.format("features_%s", entityType);
        
        Dataset<Row> features = spark.read()
            .table(featureTable)
            .select("entity_id", featureNames.toArray(new String[0]));
        
        return features;
    }
}

// Apache Spark Feature Engineering Pipeline
public class FeatureEngineeringPipeline {
    
    public Dataset<Row> engineerFeatures(SparkSession spark, Dataset<Row> rawData) {
        // 1. Data cleaning and preprocessing
        Dataset<Row> cleanedData = rawData
            .na().fill(0) // Fill missing values
            .filter(col("amount").isNotNull()) // Remove null amounts
            .filter(col("amount") > 0); // Remove invalid amounts
        
        // 2. Feature creation
        Dataset<Row> features = cleanedData
            .withColumn("amount_log", log(col("amount"))) // Log transformation
            .withColumn("amount_bucket", 
                when(col("amount") < 100, "low")
                .when(col("amount") < 1000, "medium")
                .otherwise("high")) // Categorical bucketing
            .withColumn("day_of_week", dayofweek(col("timestamp"))) // Temporal features
            .withColumn("hour_of_day", hour(col("timestamp")))
            .withColumn("is_weekend", 
                when(dayofweek(col("timestamp")).isin(1, 7), true)
                .otherwise(false));
        
        // 3. Aggregated features
        WindowSpec windowSpec = Window.partitionBy("user_id")
            .orderBy("timestamp")
            .rangeBetween(-30, -1); // Last 30 days
        
        Dataset<Row> aggregatedFeatures = features
            .withColumn("avg_amount_30d", avg("amount").over(windowSpec))
            .withColumn("count_transactions_30d", count("*").over(windowSpec))
            .withColumn("max_amount_30d", max("amount").over(windowSpec));
        
        return aggregatedFeatures;
    }
}

// Model Serving with A/B Testing
@Service
public class ModelServingService {
    
    @Autowired
    private ModelRegistry modelRegistry;
    
    @Autowired
    private FeatureStoreService featureStore;
    
    public PredictionResponse predict(String entityId, PredictionRequest request) {
        // Get model configuration
        ModelConfig config = modelRegistry.getActiveModel(request.getModelName());
        
        // Get features
        Map<String, Object> features = featureStore.getFeaturesOnline(
            entityId, config.getRequiredFeatures());
        
        // Route to appropriate model version
        ModelVersion modelVersion = routeToModelVersion(config, request);
        
        // Make prediction
        Prediction prediction = modelVersion.predict(features);
        
        // Log prediction for monitoring
        logPrediction(entityId, request, prediction, modelVersion);
        
        return new PredictionResponse(prediction, modelVersion.getVersion());
    }
    
    private ModelVersion routeToModelVersion(ModelConfig config, PredictionRequest request) {
        // A/B testing logic
        if (config.isABTestingEnabled()) {
            String userId = request.getUserId();
            int hash = Math.abs(userId.hashCode());
            
            if (hash % 100 < config.getABTestPercentage()) {
                return config.getBModelVersion(); // New model
            } else {
                return config.getAModelVersion(); // Control model
            }
        }
        
        // Canary deployment logic
        if (config.isCanaryEnabled()) {
            // Route small percentage to new model
            if (Math.random() < config.getCanaryPercentage()) {
                return config.getCanaryModelVersion();
            }
        }
        
        return config.getDefaultModelVersion();
    }
}