Lambda 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
27
28
29

# Apache Spark를 사용한 배치 처리 예시
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

# Spark 세션 초기화
spark = SparkSession.builder \
    .appName("Lambda_Batch_Layer") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# 배치 데이터 읽기 (하루치 데이터)
batch_data = spark.read \
    .option("multiline", "true") \
    .json("hdfs://cluster/raw_data/2024/01/15/")

# 배치 뷰 계산 (일일 사용자 활동 통계)
daily_stats = batch_data \
    .groupBy("user_id", "date") \
    .agg(
        count("*").alias("total_events"),
        sum("duration").alias("total_duration"),
        countDistinct("session_id").alias("sessions")
    )

# 배치 뷰 저장
daily_stats.write \
    .mode("overwrite") \
    .partitionBy("date") \
    .parquet("hdfs://cluster/batch_views/daily_user_stats/")

 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

# Apache Kafka Streams를 사용한 실시간 처리 예시
from kafka import KafkaConsumer, KafkaProducer
import json
from datetime import datetime, timedelta
import redis

# Redis 클라이언트 초기화 (실시간 뷰 저장용)
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Kafka 컨슈머 설정
consumer = KafkaConsumer(
    'user_events',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),
    group_id='lambda_speed_layer'
)

# 실시간 데이터 처리
def process_real_time_events():
    """실시간 이벤트 처리 및 뷰 업데이트"""
    for message in consumer:
        event = message.value
        user_id = event['user_id']
        timestamp = datetime.fromtimestamp(event['timestamp'])
        
        # 실시간 뷰 업데이트 (최근 1시간 활동)
        hour_key = f"user:{user_id}:hour:{timestamp.strftime('%Y%m%d%H')}"
        
        # Redis에 실시간 통계 업데이트
        redis_client.hincrby(hour_key, 'event_count', 1)
        redis_client.hincrby(hour_key, 'total_duration', event.get('duration', 0))
        redis_client.expire(hour_key, 3600)  # 1시간 TTL
        
        # 전역 실시간 통계 업데이트
        global_key = f"global:realtime:{timestamp.strftime('%Y%m%d%H%M')}"
        redis_client.hincrby(global_key, 'total_events', 1)
        redis_client.expire(global_key, 300)  # 5분 TTL

if __name__ == "__main__":
    process_real_time_events()

 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

# Flask를 사용한 통합 질의 API 예시
from flask import Flask, jsonify, request
from pyspark.sql import SparkSession
import redis
from datetime import datetime, timedelta

app = Flask(__name__)

# Spark와 Redis 클라이언트 초기화
spark = SparkSession.builder.appName("Lambda_Serving_Layer").getOrCreate()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.route('/api/user_stats/<user_id>')
def get_user_stats(user_id):
    """사용자 통계 조회 - 배치 뷰와 실시간 뷰 결합"""
    
    # 현재 시간 기준으로 배치/실시간 경계 설정
    current_time = datetime.now()
    batch_cutoff = current_time.replace(minute=0, second=0, microsecond=0)
    
    # 배치 뷰에서 과거 데이터 조회
    batch_stats = spark.sql(f"""
        SELECT 
            SUM(total_events) as batch_events,
            SUM(total_duration) as batch_duration,
            SUM(sessions) as batch_sessions
        FROM batch_views.daily_user_stats 
        WHERE user_id = '{user_id}' 
        AND date < '{batch_cutoff.strftime('%Y-%m-%d')}'
    """).collect()[0]
    
    # 실시간 뷰에서 최근 데이터 조회
    realtime_events = 0
    realtime_duration = 0
    
    # 최근 24시간의 실시간 데이터 집계
    for hour_offset in range(24):
        hour_time = current_time - timedelta(hours=hour_offset)
        hour_key = f"user:{user_id}:hour:{hour_time.strftime('%Y%m%d%H')}"
        
        hour_data = redis_client.hgetall(hour_key)
        if hour_data:
            realtime_events += int(hour_data.get(b'event_count', 0))
            realtime_duration += int(hour_data.get(b'total_duration', 0))
    
    # 배치와 실시간 결과 결합
    total_stats = {
        'user_id': user_id,
        'total_events': (batch_stats['batch_events'] or 0) + realtime_events,
        'total_duration': (batch_stats['batch_duration'] or 0) + realtime_duration,
        'total_sessions': batch_stats['batch_sessions'] or 0,
        'data_freshness': 'real-time',
        'last_updated': current_time.isoformat()
    }
    
    return jsonify(total_stats)

if __name__ == '__main__':
    app.run(debug=True, port=8080)

  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
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310

# Lambda Architecture 기반 추천 시스템 구현
import json
import redis
from datetime import datetime, timedelta
from kafka import KafkaConsumer, KafkaProducer
from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator
import numpy as np
from typing import Dict, List, Tuple

class LambdaRecommendationSystem:
    """Lambda Architecture 기반 추천 시스템"""
    
    def __init__(self):
        # Spark 세션 초기화 (배치 레이어용)
        self.spark = SparkSession.builder \
            .appName("Lambda_Recommendation_Batch") \
            .config("spark.sql.adaptive.enabled", "true") \
            .getOrCreate()
        
        # Redis 클라이언트 (실시간 데이터 저장용)
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        
        # Kafka 설정
        self.kafka_servers = ['localhost:9092']
        self.consumer = KafkaConsumer(
            'user_events',
            bootstrap_servers=self.kafka_servers,
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        self.producer = KafkaProducer(
            bootstrap_servers=self.kafka_servers,
            value_serializer=lambda x: json.dumps(x).encode('utf-8')
        )

class BatchLayer:
    """배치 레이어 - 정확한 협업 필터링 모델 학습"""
    
    def __init__(self, lambda_system: LambdaRecommendationSystem):
        self.lambda_system = lambda_system
        self.spark = lambda_system.spark
    
    def train_collaborative_filtering_model(self, data_path: str) -> None:
        """
        협업 필터링 모델 학습
        매일 밤 전체 사용자-상품 상호작용 데이터로 ALS 모델 학습
        """
        print("배치 레이어: 협업 필터링 모델 학습 시작")
        
        # 과거 상호작용 데이터 로드
        interactions_df = self.spark.read \
            .option("header", "true") \
            .option("inferSchema", "true") \
            .csv(data_path)
        
        # ALS (Alternating Least Squares) 모델 설정
        als = ALS(
            maxIter=10,
            regParam=0.1,
            userCol="user_id",
            itemCol="product_id", 
            ratingCol="rating",
            coldStartStrategy="drop",
            implicitPrefs=False
        )
        
        # 모델 학습
        model = als.fit(interactions_df)
        
        # 모델 평가
        predictions = model.transform(interactions_df)
        evaluator = RegressionEvaluator(
            metricName="rmse",
            labelCol="rating",
            predictionCol="prediction"
        )
        rmse = evaluator.evaluate(predictions)
        print(f"배치 모델 RMSE: {rmse}")
        
        # 모든 사용자에 대한 추천 생성 (Top-10)
        user_recommendations = model.recommendForAllUsers(10)
        
        # 추천 결과를 서빙 레이어에 저장
        self._save_batch_recommendations(user_recommendations)
        
        print("배치 레이어: 모델 학습 및 추천 생성 완료")
    
    def _save_batch_recommendations(self, recommendations_df) -> None:
        """배치 추천 결과를 Cassandra에 저장"""
        # 실제 환경에서는 Cassandra 커넥터 사용
        # 여기서는 Redis를 대용으로 사용
        recommendations_list = recommendations_df.collect()
        
        for row in recommendations_list:
            user_id = row.user_id
            recommendations = [
                {"product_id": rec.product_id, "score": float(rec.rating)}
                for rec in row.recommendations
            ]
            
            # Redis에 배치 추천 저장 (7일 TTL)
            batch_key = f"batch_recommendations:{user_id}"
            self.lambda_system.redis_client.setex(
                batch_key, 
                timedelta(days=7), 
                json.dumps(recommendations)
            )

class SpeedLayer:
    """스피드 레이어 - 실시간 사용자 행동 분석"""
    
    def __init__(self, lambda_system: LambdaRecommendationSystem):
        self.lambda_system = lambda_system
        self.redis_client = lambda_system.redis_client
        self.consumer = lambda_system.consumer
    
    def process_real_time_events(self) -> None:
        """실시간 사용자 이벤트 처리"""
        print("스피드 레이어: 실시간 이벤트 처리 시작")
        
        for message in self.consumer:
            event = message.value
            self._process_single_event(event)
    
    def _process_single_event(self, event: Dict) -> None:
        """개별 사용자 이벤트 처리"""
        user_id = event['user_id']
        event_type = event['event_type']
        product_id = event.get('product_id')
        timestamp = datetime.fromtimestamp(event['timestamp'])
        
        # 현재 세션 키
        session_key = f"session:{user_id}:{timestamp.strftime('%Y%m%d%H')}"
        
        if event_type == 'view':
            # 상품 조회 이벤트
            self._update_product_affinity(user_id, product_id, 1.0)
            self.redis_client.hincrby(session_key, f"view:{product_id}", 1)
            
        elif event_type == 'add_to_cart':
            # 장바구니 추가 이벤트 (더 높은 가중치)
            self._update_product_affinity(user_id, product_id, 3.0)
            self.redis_client.hincrby(session_key, f"cart:{product_id}", 1)
            
        elif event_type == 'purchase':
            # 구매 이벤트 (가장 높은 가중치)
            self._update_product_affinity(user_id, product_id, 5.0)
            self.redis_client.hincrby(session_key, f"purchase:{product_id}", 1)
        
        # 세션 만료 시간 설정 (1시간)
        self.redis_client.expire(session_key, 3600)
        
        # 실시간 추천 업데이트
        self._update_real_time_recommendations(user_id)
    
    def _update_product_affinity(self, user_id: str, product_id: str, weight: float) -> None:
        """사용자-상품 친화도 실시간 업데이트"""
        affinity_key = f"affinity:{user_id}:{product_id}"
        current_score = float(self.redis_client.get(affinity_key) or 0)
        new_score = current_score + weight
        
        # 24시간 TTL로 실시간 친화도 저장
        self.redis_client.setex(affinity_key, 86400, str(new_score))
    
    def _update_real_time_recommendations(self, user_id: str) -> None:
        """실시간 추천 업데이트"""
        # 현재 세션의 관심 상품 기반 유사 상품 추천
        # 실제로는 더 정교한 알고리즘 사용
        session_pattern = f"session:{user_id}:*"
        session_keys = self.redis_client.keys(session_pattern)
        
        recent_products = []
        for session_key in session_keys:
            session_data = self.redis_client.hgetall(session_key)
            for field, count in session_data.items():
                field_str = field.decode('utf-8') if isinstance(field, bytes) else field
                if field_str.startswith(('view:', 'cart:', 'purchase:')):
                    _, product_id = field_str.split(':', 1)
                    recent_products.append(product_id)
        
        # 최근 관심 상품 기반 실시간 추천 생성
        realtime_recommendations = self._generate_similar_products(recent_products)
        
        # Redis에 실시간 추천 저장 (1시간 TTL)
        realtime_key = f"realtime_recommendations:{user_id}"
        self.redis_client.setex(
            realtime_key, 
            3600, 
            json.dumps(realtime_recommendations)
        )
    
    def _generate_similar_products(self, product_ids: List[str]) -> List[Dict]:
        """유사 상품 추천 생성 (간단한 예시)"""
        # 실제로는 상품 임베딩, 협업 필터링 등 사용
        # 여기서는 단순화된 로직
        similar_products = []
        for product_id in product_ids[-5:]:  # 최근 5개 상품만 고려
            # 가상의 유사 상품 생성
            base_id = int(product_id) if product_id.isdigit() else hash(product_id) % 1000
            for i in range(3):  # 각 상품당 3개 유사 상품
                similar_id = str(base_id + i + 1)
                similar_products.append({
                    "product_id": similar_id,
                    "score": 0.8 - (i * 0.1),
                    "reason": f"similar_to_{product_id}"
                })
        
        # 중복 제거 및 점수순 정렬
        unique_products = {}
        for product in similar_products:
            pid = product["product_id"]
            if pid not in unique_products or product["score"] > unique_products[pid]["score"]:
                unique_products[pid] = product
        
        return sorted(unique_products.values(), key=lambda x: x["score"], reverse=True)[:10]

class ServingLayer:
    """서빙 레이어 - 배치와 실시간 추천 결합"""
    
    def __init__(self, lambda_system: LambdaRecommendationSystem):
        self.lambda_system = lambda_system
        self.redis_client = lambda_system.redis_client
    
    def get_unified_recommendations(self, user_id: str, limit: int = 10) -> List[Dict]:
        """
        통합 추천 생성
        배치 추천(장기 선호도)과 실시간 추천(즉시 관심사)을 결합
        """
        # 배치 추천 가져오기
        batch_key = f"batch_recommendations:{user_id}"
        batch_data = self.redis_client.get(batch_key)
        batch_recommendations = json.loads(batch_data) if batch_data else []
        
        # 실시간 추천 가져오기
        realtime_key = f"realtime_recommendations:{user_id}"
        realtime_data = self.redis_client.get(realtime_key)
        realtime_recommendations = json.loads(realtime_data) if realtime_data else []
        
        # 추천 결합 및 가중치 적용
        unified_recommendations = self._combine_recommendations(
            batch_recommendations, 
            realtime_recommendations
        )
        
        # 상위 N개 추천 반환
        return unified_recommendations[:limit]
    
    def _combine_recommendations(self, batch_recs: List[Dict], realtime_recs: List[Dict]) -> List[Dict]:
        """배치와 실시간 추천 결합"""
        combined = {}
        
        # 배치 추천 (가중치 0.6)
        for rec in batch_recs:
            product_id = rec["product_id"]
            combined[product_id] = {
                "product_id": product_id,
                "score": rec["score"] * 0.6,
                "sources": ["batch"]
            }
        
        # 실시간 추천 (가중치 0.4)
        for rec in realtime_recs:
            product_id = rec["product_id"]
            if product_id in combined:
                # 이미 배치 추천에 있는 경우 점수 합산
                combined[product_id]["score"] += rec["score"] * 0.4
                combined[product_id]["sources"].append("realtime")
            else:
                # 새로운 실시간 추천
                combined[product_id] = {
                    "product_id": product_id,
                    "score": rec["score"] * 0.4,
                    "sources": ["realtime"]
                }
        
        # 점수순 정렬
        return sorted(combined.values(), key=lambda x: x["score"], reverse=True)

# 사용 예시
def main():
    """Lambda Architecture 기반 추천 시스템 실행 예시"""
    
    # 시스템 초기화
    lambda_system = LambdaRecommendationSystem()
    batch_layer = BatchLayer(lambda_system)
    speed_layer = SpeedLayer(lambda_system)
    serving_layer = ServingLayer(lambda_system)
    
    # 배치 처리 (일일 실행)
    print("=== 배치 레이어 실행 ===")
    batch_layer.train_collaborative_filtering_model("hdfs://cluster/user_interactions/")
    
    # 실시간 처리 (연속 실행)
    print("=== 스피드 레이어 실행 ===")
    # 실제로는 별도 프로세스에서 실행
    # speed_layer.process_real_time_events()
    
    # 통합 추천 조회 예시
    print("=== 서빙 레이어 실행 ===")
    user_id = "user_12345"
    recommendations = serving_layer.get_unified_recommendations(user_id)
    
    print(f"사용자 {user_id}를 위한 통합 추천:")
    for i, rec in enumerate(recommendations, 1):
        print(f"{i}. 상품 {rec['product_id']} (점수: {rec['score']:f}, 소스: {rec['sources']})")

if __name__ == "__main__":
    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
34
35
36
37
38
39
40
41

# Kafka에서 실시간 스트리밍 데이터를 수신하고 Spark로 처리하는 예시
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, TimestampType

# SparkSession 생성
spark = SparkSession.builder \
    .appName("Lambda Architecture - Speed Layer") \
    .getOrCreate()

# Kafka에서 수신할 데이터 스키마 정의
schema = StructType() \
    .add("user_id", StringType()) \
    .add("event_type", StringType()) \
    .add("timestamp", TimestampType())

# Kafka 스트리밍 데이터 수신
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "click_events") \
    .load()

# Kafka의 value는 바이너리 -> 문자열로 변환
json_df = df.selectExpr("CAST(value AS STRING) as json") \
    .select(from_json("json", schema).alias("data")) \
    .select("data.*")

# 실시간 클릭 이벤트 카운팅
click_counts = json_df \
    .filter(col("event_type") == "click") \
    .groupBy("user_id") \
    .count()

# 결과를 콘솔로 출력 (Serving Layer로 저장하는 구조도 가능)
query = click_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

 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

# Kafka에서 로그 데이터를 수신하여 Spark Streaming으로 실시간 처리, Hadoop으로 배치 처리하는 예시

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

# Spark Streaming 환경 설정
sc = SparkContext(appName="LambdaArchitectureLogAnalysis")
ssc = StreamingContext(sc, 5)

# Kafka에서 로그 데이터 스트림 수신
kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181", "log-group", {"logs":1})

# 실시간 처리: 에러 로그 탐지
def detect_error(log):
    return "ERROR" in log

error_logs = kafkaStream.filter(lambda x: detect_error(x[1]))
error_logs.pprint()

# 배치 처리(예시): Hadoop MapReduce로 전체 로그 집계 (별도 배치 스크립트)
# 실제 배치 처리는 Hadoop 환경에서 별도 실행

ssc.start()
ssc.awaitTermination()