Kappa Architecture

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# event_store.py
from kafka import KafkaProducer, KafkaConsumer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def append_event(topic, event):
    producer.send(topic, event)
    producer.flush()

def replay_events(topic):
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    state = {}
    for msg in consumer:
        ev = msg.value
        # 예: entity_id 기준 상태 재구성
        state[ev['entity_id']] = ev['payload']
    return state

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# command_handler.py (쓰기 전용 RDB)
def handle_story_post(event):
    # DB에 저장 (ex: PostgreSQL)
    insert_story(event['story_id'], event['content'])

# query_service.js (읽기 전용 ElasticSearch 기반)
const { Client } = require('@elastic/elasticsearch');
const es = new Client({ node: 'http://localhost:9200' });

async function searchStories(query) {
  const resp = await es.search({
    index: 'stories',
    body: { query: { match: { content: query }}}
  });
  return resp.hits.hits;
}

1
2
3
4
5
6
7
8
9

// Kafka Streams 예시
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Long> sales = builder.stream("sales-topic", Consumed.with(Serdes.String(), Serdes.Long()));

KTable<String, Long> dailySales = sales
    .groupBy((key, value) -> key)
    .reduce(Long::sum, Materialized.as("daily-sales-store"));

dailySales.toStream().to("daily-sales-topic", Produced.with(Serdes.String(), Serdes.Long()));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.window import Tumble

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

t_env.execute_sql("""
CREATE TABLE clicks (
  user_id STRING,
  ts TIMESTAMP(3),
  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (... Kafka Source 설정 ...)
""")

t_env.execute_sql("""
CREATE TABLE click_counts (
  user_id STRING,
  cnt BIGINT,
  window_end TIMESTAMP(3)
) WITH (... Kafka Sink 설정 ...)
""")

t_env.execute_sql("""
INSERT INTO click_counts
SELECT
  user_id,
  COUNT(*) as cnt,
  TUMBLE_END(ts, INTERVAL '1' HOUR) as window_end
FROM clicks
GROUP BY
  user_id,
  TUMBLE(ts, INTERVAL '1' HOUR)
""")

1
2
3
4
5
6

# server.properties
num.partitions=12
default.replication.factor=3
compression.type=producer
log.retention.hours=168
log.segment.bytes=1073741824

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from confluent_kafka import avro, SerializingProducer
schema_str = """
{
 "namespace": "example",
 "name": "Purchase",
 "type": "record",
 "fields": [
   {"name": "user_id", "type": "string"},
   {"name": "amount", "type": "double"}
 ]
}
"""
schema = avro.loads(schema_str)

producer = SerializingProducer({
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': avro.dumps(schema),
    'value.serializer': avro.dumps(schema),
    'schema.registry.url': 'http://localhost:8081'
})

1
2
3
4

streaming:
  bufferTimeout: 100
  backpressure:
    samplingInterval: 20

1
2
3
4

# Kafka 오프셋 초기화
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group click-processor --topic click-events \
  --reset-offsets --to-earliest --execute

1
2
3
4
5
6

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.state_backend import RocksDBStateBackend

env = StreamExecutionEnvironment.get_execution_environment()
env.set_state_backend(RocksDBStateBackend("file:///flink/checkpoints", incremental=True))
env.set_parallelism(4)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# faust_app.py
import faust

# Kafka Topic 이름과 메시지 구조 정의
class TemperatureEvent(faust.Record, serializer='json'):
    device_id: str
    timestamp: str
    temperature: float

app = faust.App('kappa-iot-app', broker='kafka://localhost:9092')

# 입력 토픽
temp_topic = app.topic('iot-temp', value_type=TemperatureEvent)

# 테이블 (materialized view 역할)
alert_store = app.Table('overheat-events', default=int)

# 스트림 처리 로직
@app.agent(temp_topic)
async def process_temperature(stream):
    async for event in stream:
        if event.temperature > 75.0:
            alert_store[event.device_id] += 1
            print(f"[ALERT] {event.device_id} temperature {event.temperature} exceeds threshold!")
        else:
            print(f"{event.device_id}: temperature is normal ({event.temperature})")

if __name__ == '__main__':
    app.main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# Kafka에서 거래 데이터 스트림을 수신하고 Flink로 실시간 처리하는 예시

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.common.serialization import SimpleStringSchema

# 환경 설정
env = StreamExecutionEnvironment.get_execution_environment()

# Kafka Consumer 생성
kafka_consumer = FlinkKafkaConsumer(
    topics='transactions',
    deserialization_schema=SimpleStringSchema(),
    properties={'bootstrap.servers': 'localhost:9092', 'group.id': 'monitoring-group'}
)

# 데이터 스트림 생성
stream = env.add_source(kafka_consumer)

# 실시간 이상 거래 탐지 로직 (예시)
def detect_fraud(transaction):
    # 거래 금액이 1000만원 이상이면 이상 거래로 간주
    if transaction.amount > 10000000:
        return True
    return False

# 이상 거래 필터링
fraud_stream = stream.filter(detect_fraud)

# 결과 출력 또는 저장
fraud_stream.print()

env.execute("Real-Time Transaction Monitoring")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252

# Kafka Consumer를 활용한 실시간 이벤트 처리
from kafka import KafkaConsumer, KafkaProducer
import json
import redis
import pandas as pd
from datetime import datetime
import logging

class KappaRecommendationEngine:
    """
    Kappa Architecture 기반 실시간 추천 엔진
    단일 스트림 처리로 사용자 행동을 실시간 분석하여 추천 생성
    """
    
    def __init__(self):
        # Kafka Consumer 설정 - 사용자 이벤트 스트림 수신
        self.consumer = KafkaConsumer(
            'user-events',  # 사용자 행동 이벤트 토픽
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda x: json.loads(x.decode('utf-8')),
            group_id='recommendation-engine'  # 컨슈머 그룹 설정
        )
        
        # Kafka Producer 설정 - 추천 결과 스트림 전송
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda x: json.dumps(x).encode('utf-8')
        )
        
        # Redis 캐시 설정 - 실시간 추천 결과 저장
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        
        # 사용자 프로필 캐시 (메모리 기반 상태 관리)
        self.user_profiles = {}
        
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def process_event_stream(self):
        """
        실시간 이벤트 스트림 처리
        Kappa Architecture의 핵심 - 모든 데이터를 스트림으로 처리
        """
        for message in self.consumer:
            try:
                event = message.value
                self.logger.info(f"Processing event: {event['event_type']}")
                
                # 이벤트 타입별 처리 (스트림 처리 로직)
                if event['event_type'] == 'view':
                    self._handle_view_event(event)
                elif event['event_type'] == 'rating':
                    self._handle_rating_event(event)
                elif event['event_type'] == 'search':
                    self._handle_search_event(event)
                
                # 실시간 추천 생성 및 업데이트
                recommendations = self._generate_recommendations(event['user_id'])
                self._update_recommendations(event['user_id'], recommendations)
                
            except Exception as e:
                self.logger.error(f"Error processing event: {e}")
    
    def _handle_view_event(self, event):
        """
        시청 이벤트 처리 - 사용자 선호도 실시간 업데이트
        """
        user_id = event['user_id']
        content_id = event['content_id']
        watch_duration = event.get('watch_duration', 0)
        
        # 사용자 프로필 업데이트 (상태 관리)
        if user_id not in self.user_profiles:
            self.user_profiles[user_id] = {
                'viewed_content': [],
                'preferences': {},
                'last_activity': None
            }
        
        # 시청 기록 추가
        self.user_profiles[user_id]['viewed_content'].append({
            'content_id': content_id,
            'timestamp': datetime.now().isoformat(),
            'watch_duration': watch_duration
        })
        
        # 선호도 점수 계산 (시청 시간 기반)
        preference_score = min(watch_duration / 3600, 1.0)  # 최대 1시간 기준 정규화
        genre = self._get_content_genre(content_id)
        
        if genre in self.user_profiles[user_id]['preferences']:
            # 기존 선호도와 가중 평균
            current_score = self.user_profiles[user_id]['preferences'][genre]
            new_score = (current_score * 0.8) + (preference_score * 0.2)
            self.user_profiles[user_id]['preferences'][genre] = new_score
        else:
            self.user_profiles[user_id]['preferences'][genre] = preference_score
    
    def _handle_rating_event(self, event):
        """
        평점 이벤트 처리 - 명시적 피드백 반영
        """
        user_id = event['user_id']
        content_id = event['content_id']
        rating = event['rating']
        
        # 평점 기반 선호도 조정
        genre = self._get_content_genre(content_id)
        rating_score = (rating - 2.5) / 2.5  # -1 ~ 1 범위로 정규화
        
        if user_id in self.user_profiles:
            if genre in self.user_profiles[user_id]['preferences']:
                current_score = self.user_profiles[user_id]['preferences'][genre]
                # 평점은 더 높은 가중치 적용
                new_score = (current_score * 0.6) + (rating_score * 0.4)
                self.user_profiles[user_id]['preferences'][genre] = max(-1, min(1, new_score))
    
    def _generate_recommendations(self, user_id):
        """
        실시간 추천 생성 - 협업 필터링 + 콘텐츠 기반 필터링
        """
        if user_id not in self.user_profiles:
            return self._get_popular_content()  # 신규 사용자 대상 인기 콘텐츠
        
        user_preferences = self.user_profiles[user_id]['preferences']
        viewed_content = [item['content_id'] for item in self.user_profiles[user_id]['viewed_content']]
        
        # 선호 장르 기반 콘텐츠 추천
        recommendations = []
        
        # 상위 선호 장르 선택
        top_genres = sorted(user_preferences.items(), key=lambda x: x[1], reverse=True)[:3]
        
        for genre, preference_score in top_genres:
            if preference_score > 0.3:  # 최소 선호도 기준
                genre_content = self._get_content_by_genre(genre)
                # 이미 본 콘텐츠 제외
                new_content = [c for c in genre_content if c not in viewed_content]
                recommendations.extend(new_content[:5])  # 장르별 5개까지
        
        # 유사 사용자 기반 추천 (협업 필터링)
        similar_users = self._find_similar_users(user_id)
        for similar_user in similar_users[:3]:  # 상위 3명의 유사 사용자
            similar_content = [item['content_id'] for item in self.user_profiles[similar_user]['viewed_content']]
            new_content = [c for c in similar_content if c not in viewed_content and c not in recommendations]
            recommendations.extend(new_content[:3])  # 유사 사용자별 3개까지
        
        return recommendations[:10]  # 최대 10개 추천
    
    def _update_recommendations(self, user_id, recommendations):
        """
        추천 결과를 Redis에 캐시하고 추천 스트림으로 전송
        """
        # Redis에 추천 결과 저장 (TTL 설정)
        cache_key = f"recommendations:{user_id}"
        recommendation_data = {
            'user_id': user_id,
            'recommendations': recommendations,
            'timestamp': datetime.now().isoformat()
        }
        
        self.redis_client.setex(
            cache_key,
            3600,  # 1시간 TTL
            json.dumps(recommendation_data)
        )
        
        # Kafka로 추천 결과 스트림 전송
        self.producer.send('recommendations', recommendation_data)
        
        self.logger.info(f"Updated recommendations for user {user_id}: {len(recommendations)} items")
    
    def _get_content_genre(self, content_id):
        """
        콘텐츠 장르 조회 (실제 구현에서는 외부 메타데이터 서비스 호출)
        """
        # 간단한 시뮬레이션
        genres = ['action', 'comedy', 'drama', 'thriller', 'documentary']
        return genres[hash(content_id) % len(genres)]
    
    def _get_content_by_genre(self, genre):
        """
        장르별 콘텐츠 조회
        """
        # 실제 구현에서는 콘텐츠 카탈로그 서비스 호출
        return [f"{genre}_content_{i}" for i in range(1, 21)]
    
    def _find_similar_users(self, user_id):
        """
        유사 사용자 찾기 - 코사인 유사도 기반
        """
        if user_id not in self.user_profiles:
            return []
        
        target_preferences = self.user_profiles[user_id]['preferences']
        similar_users = []
        
        for other_user_id, profile in self.user_profiles.items():
            if other_user_id == user_id:
                continue
            
            # 코사인 유사도 계산
            similarity = self._calculate_cosine_similarity(
                target_preferences, 
                profile['preferences']
            )
            
            if similarity > 0.7:  # 유사도 임계값
                similar_users.append((other_user_id, similarity))
        
        # 유사도 순으로 정렬
        similar_users.sort(key=lambda x: x[1], reverse=True)
        return [user_id for user_id, _ in similar_users]
    
    def _calculate_cosine_similarity(self, prefs1, prefs2):
        """
        코사인 유사도 계산
        """
        # 공통 장르 찾기
        common_genres = set(prefs1.keys()) & set(prefs2.keys())
        if not common_genres:
            return 0
        
        # 벡터 내적과 크기 계산
        dot_product = sum(prefs1[genre] * prefs2[genre] for genre in common_genres)
        norm1 = sum(prefs1[genre] ** 2 for genre in common_genres) ** 0.5
        norm2 = sum(prefs2[genre] ** 2 for genre in common_genres) ** 0.5
        
        if norm1 == 0 or norm2 == 0:
            return 0
        
        return dot_product / (norm1 * norm2)
    
    def _get_popular_content(self):
        """
        인기 콘텐츠 조회 (신규 사용자용)
        """
        return [f"popular_content_{i}" for i in range(1, 11)]

# 실행 예시
if __name__ == "__main__":
    """
    Kappa Architecture 추천 엔진 실행
    모든 사용자 이벤트를 실시간 스트림으로 처리하여 즉시 추천 업데이트
    """
    engine = KappaRecommendationEngine()
    
    # 실시간 이벤트 스트림 처리 시작
    print("Starting Kappa Recommendation Engine…")
    print("Processing real-time user events…")
    
    engine.process_event_stream()