大数据架构设计：Lambda架构与Kappa架构对比

    class LambdaMerger:
    def __init__(self, batch_view, realtime_view):
        self.batch = batch_view  # 批处理结果集
        self.realtime = realtime_view  # 实时处理结果集
        
    def query(self, key):
        # 优先使用批处理数据
        base = self.batch.get(key)  
        # 补充实时增量
        delta = self.realtime.get(key)  
        return self._merge(base, delta)
    
    def _merge(self, base, delta):
        if base is None:
            return delta
        if delta is None:
            return base
        # 使用时间戳解决冲突
        return max(base, delta, key=lambda x: x['timestamp'])
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/aHsRvltO3A9YNEwmMPLK41D2CjUV.png)

    class KappaProcessor:
    def __init__(self, kafka_topic):
        self.consumer = KafkaConsumer(kafka_topic)
        self.state = {}
        
    def process_stream(self):
        for message in self.consumer:
            event = self._parse(message)
            self._update_state(event)
    
    def _update_state(self, event):
        key = event['user_id']
        current = self.state.get(key, {'count':0})
        current['count'] += 1
        self.state[key] = current
        
    def replay_events(self, start_offset):
        self.state.clear()
        self.consumer.seek(start_offset)
        self.process_stream()
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/XLDbhZjG3MHgpseUWA78t0cTwf6S.png)

    # Lambda架构技术栈
    docker run -d --name hadoop hadoop:3.3
    docker run -d --name flink flink:1.15
    docker run -d --name druid druid:0.23
    
    # Kappa架构技术栈
    docker run -d --name kafka kafka:3.2
    docker run -d --name flink flink:1.15
    
    
    bash

    val batchView = spark.read.parquet("/data/warehouse")
      .groupBy("user_id")
      .agg(count("*").as("total_orders"))
    
    
    scala

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.addSource(new FlinkKafkaConsumer<>("orders", new JSONDeserializer(), properties))
      .keyBy(event -> event.userId)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
      .aggregate(new OrderCounter());
    
    
    java