Enterprise Data Pipeline Design

// Speed Layer Implementation with Apache Flink
public class SpeedLayerProcessor {
    private final StreamExecutionEnvironment env;
    
    public SpeedLayerProcessor() {
        this.env = StreamExecutionEnvironment.getExecutionEnvironment();
        this.env.enableCheckpointing(60000); // 1 minute checkpoints
    }
    
    public void processRealTimeData(DataStream<Event> eventStream) {
        eventStream
            .keyBy(Event::getUserId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .aggregate(new RealTimeAggregator())
            .addSink(new SpeedLayerSink());
    }
}

// Batch Layer Implementation with Apache Spark
public class BatchLayerProcessor {
    private final SparkSession spark;
    
    public BatchLayerProcessor() {
        this.spark = SparkSession.builder()
            .appName("Lambda-Batch-Layer")
            .config("spark.sql.adaptive.enabled", "true")
            .getOrCreate();
    }
    
    public Dataset<Row> processBatchData(String dataPath) {
        return spark.read()
            .option("header", "true")
            .csv(dataPath)
            .transform(new BatchDataTransformer())
            .transform(new DataQualityValidator())
            .transform(new MasterDatasetBuilder());
    }
}

// Serving Layer Implementation
public class ServingLayer {
    private final SpeedLayerClient speedClient;
    private final BatchLayerClient batchClient;
    
    public ServingLayer() {
        this.speedClient = new SpeedLayerClient();
        this.batchClient = new BatchLayerClient();
    }
    
    public CompletableFuture<AggregatedResult> getUnifiedResult(String userId) {
        CompletableFuture<SpeedResult> speedResult = speedClient.getLatestResult(userId);
        CompletableFuture<BatchResult> batchResult = batchClient.getMasterData(userId);
        
        return CompletableFuture.allOf(speedResult, batchResult)
            .thenApply(v -> mergeResults(speedResult.join(), batchResult.join()));
    }
    
    private AggregatedResult mergeResults(SpeedResult speed, BatchResult batch) {
        // Merge logic: speed results override batch results for recent data
        return new AggregatedResult(speed, batch, Instant.now());
    }
}

// Kappa Architecture Stream Processor
public class KappaStreamProcessor {
    private final StreamExecutionEnvironment env;
    private final KafkaSource<String> source;
    
    public KappaStreamProcessor() {
        this.env = StreamExecutionEnvironment.getExecutionEnvironment();
        this.source = KafkaSource.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("data-stream")
            .setGroupId("kappa-processor")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .build();
    }
    
    public void processStream() {
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
            .map(new JsonDeserializer())
            .keyBy(Event::getEntityId)
            .process(new StatefulEventProcessor())
            .addSink(new ResultSink());
    }
}

// State Management in Kappa Architecture
public class StatefulEventProcessor extends KeyedProcessFunction<String, Event, ProcessedEvent> {
    private ValueState<EntityState> entityState;
    private ListState<Event> eventHistory;
    
    @Override
    public void open(Configuration parameters) {
        entityState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("entity-state", EntityState.class)
        );
        eventHistory = getRuntimeContext().getListState(
            new ListStateDescriptor<>("event-history", Event.class)
        );
    }
    
    @Override
    public void processElement(Event event, Context ctx, Collector<ProcessedEvent> out) throws Exception {
        EntityState currentState = entityState.value();
        if (currentState == null) {
            currentState = new EntityState();
        }
        
        // Update state based on event
        currentState = updateState(currentState, event);
        entityState.update(currentState);
        
        // Store event in history
        eventHistory.add(event);
        
        // Emit processed result
        out.collect(new ProcessedEvent(event.getId(), currentState, Instant.now()));
    }
}

// Stream Replay Implementation
public class StreamReplayService {
    private final KafkaConsumer<String, String> consumer;
    private final StreamExecutionEnvironment env;
    
    public void replayFromTimestamp(Instant startTime, Instant endTime) {
        // Configure consumer for replay
        consumer.seekToBeginning(Collections.singletonList(topicPartition));
        
        // Create replay stream
        DataStream<Event> replayStream = env
            .addSource(new ReplaySource(startTime, endTime))
            .filter(event -> isInTimeRange(event, startTime, endTime))
            .keyBy(Event::getEntityId)
            .process(new ReplayProcessor());
        
        // Process replay stream
        replayStream.addSink(new ReplayResultSink());
    }
    
    private boolean isInTimeRange(Event event, Instant start, Instant end) {
        Instant eventTime = event.getTimestamp();
        return !eventTime.isBefore(start) && !eventTime.isAfter(end);
    }
}