Data-Flow Architecture

1
2
3
4
5
6
7

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("daily_etl").getOrCreate()
df = spark.read.csv("s3://bucket/raw/transactions.csv", header=True, inferSchema=True)
df_clean = df.filter(df['amount'] > 0).select("id","amount","timestamp")
df_clean.write.mode("overwrite").parquet("s3://bucket/processed/transactions/")
spark.stop()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "flink-group");
DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("input-topic", new SimpleStringSchema(), props));
DataStream<String> processed = stream
    .filter(s -> s.contains("error"))
    .map(s -> s.toUpperCase());
processed.addSink(new FlinkKafkaProducer<>("output-topic", new SimpleStringSchema(), props));
env.execute("Kafka-Flink Stream");

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.utils.dates import days_ago

dag = DAG('etl_dag', start_date=days_ago(1), schedule_interval="@daily")

spark_task = SparkSubmitOperator(
    task_id="spark_etl",
    application="/scripts/daily_etl.py",
    conn_id="spark_default",
    dag=dag
)

  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

import asyncio
import json
from abc import ABC, abstractmethod
from typing import Any, Dict, List, AsyncIterator
from dataclasses import dataclass
from datetime import datetime

# 기본 필터 인터페이스 정의
class Filter(ABC):
    """Data-Flow Architecture의 기본 필터 인터페이스"""
    
    @abstractmethod
    async def process(self, data: Any) -> Any:
        """데이터 처리 메서드"""
        pass

# 파이프 구현 - 필터 간 데이터 전송
class Pipe:
    """필터 간 데이터 전송을 담당하는 파이프"""
    
    def __init__(self, buffer_size: int = 1000):
        self.queue = asyncio.Queue(maxsize=buffer_size)
    
    async def send(self, data: Any):
        """데이터 전송"""
        await self.queue.put(data)
    
    async def receive(self) -> Any:
        """데이터 수신"""
        return await self.queue.get()

# 사용자 이벤트 데이터 구조
@dataclass
class UserEvent:
    user_id: str
    content_id: str
    event_type: str  # 'view', 'click', 'rate'
    timestamp: datetime
    duration: float = 0.0
    rating: float = 0.0

# 1. 이벤트 수집 필터
class EventCollectorFilter(Filter):
    """사용자 이벤트를 수집하고 정규화하는 필터"""
    
    async def process(self, raw_event: Dict[str, Any]) -> UserEvent:
        """원시 이벤트를 UserEvent 객체로 변환"""
        try:
            event = UserEvent(
                user_id=raw_event['user_id'],
                content_id=raw_event['content_id'],
                event_type=raw_event['event_type'],
                timestamp=datetime.fromisoformat(raw_event['timestamp']),
                duration=raw_event.get('duration', 0.0),
                rating=raw_event.get('rating', 0.0)
            )
            print(f"이벤트 수집: {event.user_id} - {event.event_type}")
            return event
        except Exception as e:
            print(f"이벤트 처리 오류: {e}")
            return None

# 2. 특성 추출 필터
class FeatureExtractionFilter(Filter):
    """사용자 행동에서 특성을 추출하는 필터"""
    
    def __init__(self):
        self.user_profiles = {}  # 실제로는 Redis 등 외부 저장소 사용
    
    async def process(self, event: UserEvent) -> Dict[str, Any]:
        """사용자 이벤트에서 특성 추출"""
        if not event:
            return None
        
        # 사용자 프로필 업데이트
        if event.user_id not in self.user_profiles:
            self.user_profiles[event.user_id] = {
                'view_count': 0,
                'total_duration': 0.0,
                'preferred_genres': {},
                'recent_content': []
            }
        
        profile = self.user_profiles[event.user_id]
        
        if event.event_type == 'view':
            profile['view_count'] += 1
            profile['total_duration'] += event.duration
            profile['recent_content'].append(event.content_id)
            
            # 최근 10개 콘텐츠만 유지
            if len(profile['recent_content']) > 10:
                profile['recent_content'] = profile['recent_content'][-10:]
        
        features = {
            'user_id': event.user_id,
            'avg_watch_duration': profile['total_duration'] / max(profile['view_count'], 1),
            'activity_level': 'high' if profile['view_count'] > 50 else 'medium' if profile['view_count'] > 10 else 'low',
            'recent_content': profile['recent_content']
        }
        
        print(f"특성 추출 완료: {event.user_id} - 활동 레벨: {features['activity_level']}")
        return features

# 3. 추천 생성 필터
class RecommendationFilter(Filter):
    """ML 모델을 사용하여 추천을 생성하는 필터"""
    
    def __init__(self):
        # 실제로는 TensorFlow, PyTorch 등의 모델 로드
        self.content_catalog = {
            'movie_1': {'genre': 'action', 'rating': 4.5},
            'movie_2': {'genre': 'comedy', 'rating': 4.2},
            'movie_3': {'genre': 'drama', 'rating': 4.7},
            'movie_4': {'genre': 'action', 'rating': 4.3},
            'movie_5': {'genre': 'comedy', 'rating': 4.0}
        }
    
    async def process(self, features: Dict[str, Any]) -> Dict[str, Any]:
        """사용자 특성을 기반으로 추천 생성"""
        if not features:
            return None
        
        user_id = features['user_id']
        activity_level = features['activity_level']
        recent_content = features['recent_content']
        
        # 간단한 추천 로직 (실제로는 복잡한 ML 모델 사용)
        recommendations = []
        
        # 활동 레벨에 따른 추천 수 조정
        rec_count = {'high': 10, 'medium': 5, 'low': 3}[activity_level]
        
        # 최근 시청하지 않은 콘텐츠 추천
        for content_id, content_info in self.content_catalog.items():
            if content_id not in recent_content:
                recommendations.append({
                    'content_id': content_id,
                    'score': content_info['rating'],
                    'genre': content_info['genre']
                })
        
        # 점수 기준 정렬 및 상위 N개 선택
        recommendations.sort(key=lambda x: x['score'], reverse=True)
        recommendations = recommendations[:rec_count]
        
        result = {
            'user_id': user_id,
            'recommendations': recommendations,
            'generated_at': datetime.now().isoformat()
        }
        
        print(f"추천 생성 완료: {user_id} - {len(recommendations)}개 추천")
        return result

# 4. 캐시 필터
class CacheFilter(Filter):
    """추천 결과를 캐시에 저장하는 필터"""
    
    def __init__(self):
        self.cache = {}  # 실제로는 Redis 등 사용
    
    async def process(self, recommendation: Dict[str, Any]) -> Dict[str, Any]:
        """추천 결과를 캐시에 저장"""
        if not recommendation:
            return None
        
        user_id = recommendation['user_id']
        self.cache[user_id] = recommendation
        
        print(f"캐시 저장 완료: {user_id}")
        return recommendation

# Data-Flow 파이프라인 구성
class RecommendationPipeline:
    """Netflix 스타일 추천 시스템 파이프라인"""
    
    def __init__(self):
        # 필터 인스턴스 생성
        self.event_collector = EventCollectorFilter()
        self.feature_extractor = FeatureExtractionFilter()
        self.recommender = RecommendationFilter()
        self.cache = CacheFilter()
        
        # 파이프 생성
        self.pipe1 = Pipe()  # EventCollector -> FeatureExtractor
        self.pipe2 = Pipe()  # FeatureExtractor -> Recommender
        self.pipe3 = Pipe()  # Recommender -> Cache
    
    async def process_event(self, raw_event: Dict[str, Any]):
        """단일 이벤트 처리"""
        # 1단계: 이벤트 수집 및 정규화
        event = await self.event_collector.process(raw_event)
        if event:
            await self.pipe1.send(event)
    
    async def run_pipeline(self):
        """파이프라인 실행 - 각 필터를 독립적인 태스크로 실행"""
        
        # 특성 추출 태스크
        async def feature_extraction_task():
            while True:
                try:
                    event = await self.pipe1.receive()
                    features = await self.feature_extractor.process(event)
                    if features:
                        await self.pipe2.send(features)
                except Exception as e:
                    print(f"특성 추출 오류: {e}")
        
        # 추천 생성 태스크
        async def recommendation_task():
            while True:
                try:
                    features = await self.pipe2.receive()
                    recommendation = await self.recommender.process(features)
                    if recommendation:
                        await self.pipe3.send(recommendation)
                except Exception as e:
                    print(f"추천 생성 오류: {e}")
        
        # 캐시 저장 태스크
        async def cache_task():
            while True:
                try:
                    recommendation = await self.pipe3.receive()
                    await self.cache.process(recommendation)
                except Exception as e:
                    print(f"캐시 저장 오류: {e}")
        
        # 모든 태스크를 동시에 실행
        await asyncio.gather(
            feature_extraction_task(),
            recommendation_task(),
            cache_task()
        )

# 사용 예시
async def main():
    """메인 실행 함수"""
    pipeline = RecommendationPipeline()
    
    # 파이프라인 시작
    pipeline_task = asyncio.create_task(pipeline.run_pipeline())
    
    # 테스트 이벤트 생성
    test_events = [
        {
            'user_id': 'user_123',
            'content_id': 'movie_1',
            'event_type': 'view',
            'timestamp': datetime.now().isoformat(),
            'duration': 1200.0
        },
        {
            'user_id': 'user_123',
            'content_id': 'movie_2',
            'event_type': 'click',
            'timestamp': datetime.now().isoformat()
        },
        {
            'user_id': 'user_456',
            'content_id': 'movie_3',
            'event_type': 'view',
            'timestamp': datetime.now().isoformat(),
            'duration': 3600.0
        }
    ]
    
    # 이벤트 처리
    for event in test_events:
        await pipeline.process_event(event)
        await asyncio.sleep(0.1)  # 약간의 지연
    
    # 결과 확인을 위한 대기
    await asyncio.sleep(2)
    
    print("=== 캐시된 추천 결과 ===")
    for user_id, recommendation in pipeline.cache.cache.items():
        print(f"사용자 {user_id}:")
        for rec in recommendation['recommendations'][:3]:  # 상위 3개만 출력
            print(f"  - {rec['content_id']} (점수: {rec['score']}, 장르: {rec['genre']})")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

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

from kafka import KafkaConsumer, KafkaProducer
from pyspark.sql import SparkSession

# 실시간 스트림 처리 (Flink 유사 PySpark Structured Streaming 예시)
spark = SparkSession.builder.appName("stream_app").getOrCreate()
df = spark.readStream.format("kafka")\
   .option("subscribe", "sensor")\
   .load()
processed = df.selectExpr("CAST(value AS STRING) as raw")\
   .filter("raw like '%alarm%'")
processed.writeStream.format("console").start()

# 배치 처리 부분 (Pyspark 예시)
batch_spark = SparkSession.builder.appName("batch_app").getOrCreate()
batch_df = batch_spark.read.json("/data/sensor/*.json")
agg = batch_df.groupBy("device_id").count()
agg.write.mode("overwrite").parquet("/data/daily_counts")

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

# 필터1: 데이터 전처리
def filter1(data):
    return data.lower()

# 필터2: 데이터 분석
def filter2(data):
    return data.replace("error", "warning")

# 필터3: 데이터 출력
def filter3(data):
    print("결과:", data)

# 파이프라인 실행
data = "This is an ERROR message"
result = filter1(data)
result = filter2(result)
filter3(result)