Lambda Architecture

// Lambda Architecture Configuration
@Configuration
public class LambdaArchitectureConfig {
    
    @Bean
    public BatchProcessor batchProcessor() {
        return BatchProcessor.builder()
            .batchSize(10000)
            .processingInterval(Duration.ofHours(1))
            .dataSource("historical-data")
            .build();
    }
    
    @Bean
    public SpeedProcessor speedProcessor() {
        return SpeedProcessor.builder()
            .windowSize(Duration.ofMinutes(5))
            .processingInterval(Duration.ofSeconds(30))
            .dataSource("real-time-stream")
            .build();
    }
    
    @Bean
    public ServingLayer servingLayer() {
        return ServingLayer.builder()
            .batchResults("batch-results")
            .speedResults("speed-results")
            .mergeStrategy(MergeStrategy.LATEST_WINS)
            .build();
    }
}

// Batch Processing Implementation
@Component
public class BatchProcessor {
    
    @Autowired
    private SparkSession sparkSession;
    
    public void processBatchData(String date) {
        // Read historical data
        Dataset<Row> historicalData = sparkSession.read()
            .option("basePath", "/data/historical")
            .parquet("/data/historical/*");
        
        // Apply batch transformations
        Dataset<Row> processedData = historicalData
            .filter(col("date").leq(date))
            .groupBy("user_id", "category")
            .agg(
                sum("amount").as("total_amount"),
                count("*").as("transaction_count"),
                avg("amount").as("avg_amount")
            );
        
        // Write results to serving layer
        processedData.write()
            .mode(SaveMode.Overwrite)
            .partitionBy("date")
            .parquet("/data/batch-results/" + date);
    }
}