Data-Centric Architecture

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

# Kafka 소비자 예시: 고객 행동 수집
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'customer_behavior',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    data = message.value
    customer_id = data.get("user_id")
    event = data.get("action_type")
    timestamp = data.get("timestamp")
    print(f"[{timestamp}] User {customer_id} performed {event}")

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

# 고객 통합 (CDP-like ID mapping)
def unify_customer_records(records):
    unified = {}
    for record in records:
        email = record.get('email')
        phone = record.get('phone')
        key = email or phone
        if key not in unified:
            unified[key] = record
        else:
            unified[key].update(record)
    return unified

  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
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363

"""
Data-Centric Architecture 구현 예시
Repository Pattern과 Data Access Layer를 중심으로 구현
"""
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
import asyncio
from dataclasses import dataclass, asdict

# 도메인 모델 정의
@dataclass
class Customer:
    """고객 도메인 모델 - 데이터 중심 설계의 핵심 엔티티"""
    id: str
    name: str
    email: str
    created_at: datetime
    last_updated: datetime
    
    def to_dict(self) -> Dict[str, Any]:
        """딕셔너리 변환 - 데이터 교환 표준화"""
        return asdict(self)

@dataclass
class Order:
    """주문 도메인 모델"""
    id: str
    customer_id: str
    amount: float
    status: str
    created_at: datetime
    
    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

# Repository Pattern 구현 - 데이터 접근 추상화
class Repository(ABC):
    """추상 저장소 인터페이스 - 데이터 접근 방식 표준화"""
    
    @abstractmethod
    async def create(self, entity: Any) -> str:
        """엔티티 생성"""
        pass
    
    @abstractmethod
    async def get_by_id(self, entity_id: str) -> Optional[Any]:
        """ID로 엔티티 조회"""
        pass
    
    @abstractmethod
    async def update(self, entity: Any) -> bool:
        """엔티티 업데이트"""
        pass
    
    @abstractmethod
    async def delete(self, entity_id: str) -> bool:
        """엔티티 삭제"""
        pass

class CustomerRepository(Repository):
    """고객 저장소 구현 - 공유 데이터 모델 활용"""
    
    def __init__(self, data_access_layer):
        self.dal = data_access_layer
        self.collection_name = "customers"
    
    async def create(self, customer: Customer) -> str:
        """고객 데이터 생성 - 표준화된 접근"""
        data = customer.to_dict()
        return await self.dal.insert(self.collection_name, data)
    
    async def get_by_id(self, customer_id: str) -> Optional[Customer]:
        """고객 ID로 조회"""
        data = await self.dal.find_by_id(self.collection_name, customer_id)
        if data:
            return Customer(**data)
        return None
    
    async def get_by_email(self, email: str) -> Optional[Customer]:
        """이메일로 고객 조회 - 비즈니스 로직 캡슐화"""
        data = await self.dal.find_by_field(self.collection_name, "email", email)
        if data:
            return Customer(**data)
        return None
    
    async def update(self, customer: Customer) -> bool:
        """고객 정보 업데이트"""
        customer.last_updated = datetime.now()
        data = customer.to_dict()
        return await self.dal.update(self.collection_name, customer.id, data)
    
    async def delete(self, customer_id: str) -> bool:
        """고객 삭제"""
        return await self.dal.delete(self.collection_name, customer_id)

class OrderRepository(Repository):
    """주문 저장소 구현"""
    
    def __init__(self, data_access_layer):
        self.dal = data_access_layer
        self.collection_name = "orders"
    
    async def create(self, order: Order) -> str:
        data = order.to_dict()
        return await self.dal.insert(self.collection_name, data)
    
    async def get_by_id(self, order_id: str) -> Optional[Order]:
        data = await self.dal.find_by_id(self.collection_name, order_id)
        if data:
            return Order(**data)
        return None
    
    async def get_by_customer_id(self, customer_id: str) -> List[Order]:
        """고객별 주문 목록 조회 - 관계형 데이터 처리"""
        data_list = await self.dal.find_by_field(self.collection_name, "customer_id", customer_id)
        return [Order(**data) for data in data_list]
    
    async def update(self, order: Order) -> bool:
        data = order.to_dict()
        return await self.dal.update(self.collection_name, order.id, data)
    
    async def delete(self, order_id: str) -> bool:
        return await self.dal.delete(self.collection_name, order_id)

# Data Access Layer - 데이터 접근 통합 계층
class DataAccessLayer:
    """데이터 접근 계층 - 다양한 데이터 소스 통합"""
    
    def __init__(self):
        # 실제 구현에서는 데이터베이스 연결 설정
        self._storage = {}  # 메모리 저장소 (데모용)
        self._metadata = {}  # 메타데이터 저장
        self._event_handlers = []  # 이벤트 핸들러
    
    async def insert(self, collection: str, data: Dict[str, Any]) -> str:
        """데이터 삽입 - 표준화된 인터페이스"""
        if collection not in self._storage:
            self._storage[collection] = {}
        
        entity_id = data.get('id', str(len(self._storage[collection]) + 1))
        self._storage[collection][entity_id] = data
        
        # 메타데이터 업데이트
        await self._update_metadata(collection, 'insert', entity_id)
        
        # 이벤트 발행 - 데이터 변경 알림
        await self._publish_event('data_inserted', collection, entity_id, data)
        
        return entity_id
    
    async def find_by_id(self, collection: str, entity_id: str) -> Optional[Dict[str, Any]]:
        """ID로 데이터 조회"""
        if collection in self._storage and entity_id in self._storage[collection]:
            return self._storage[collection][entity_id]
        return None
    
    async def find_by_field(self, collection: str, field: str, value: Any) -> List[Dict[str, Any]]:
        """필드 값으로 데이터 조회 - 유연한 쿼리 지원"""
        results = []
        if collection in self._storage:
            for data in self._storage[collection].values():
                if data.get(field) == value:
                    results.append(data)
        return results
    
    async def update(self, collection: str, entity_id: str, data: Dict[str, Any]) -> bool:
        """데이터 업데이트"""
        if collection in self._storage and entity_id in self._storage[collection]:
            old_data = self._storage[collection][entity_id].copy()
            self._storage[collection][entity_id] = data
            
            await self._update_metadata(collection, 'update', entity_id)
            await self._publish_event('data_updated', collection, entity_id, data, old_data)
            return True
        return False
    
    async def delete(self, collection: str, entity_id: str) -> bool:
        """데이터 삭제"""
        if collection in self._storage and entity_id in self._storage[collection]:
            deleted_data = self._storage[collection].pop(entity_id)
            
            await self._update_metadata(collection, 'delete', entity_id)
            await self._publish_event('data_deleted', collection, entity_id, deleted_data)
            return True
        return False
    
    async def _update_metadata(self, collection: str, operation: str, entity_id: str):
        """메타데이터 업데이트 - 데이터 계보 추적"""
        if collection not in self._metadata:
            self._metadata[collection] = {
                'total_records': 0,
                'operations': [],
                'last_updated': None
            }
        
        self._metadata[collection]['operations'].append({
            'operation': operation,
            'entity_id': entity_id,
            'timestamp': datetime.now().isoformat()
        })
        self._metadata[collection]['last_updated'] = datetime.now().isoformat()
        
        if operation in ['insert']:
            self._metadata[collection]['total_records'] += 1
        elif operation in ['delete']:
            self._metadata[collection]['total_records'] -= 1
    
    async def _publish_event(self, event_type: str, collection: str, entity_id: str, 
                           data: Dict[str, Any], old_data: Dict[str, Any] = None):
        """이벤트 발행 - 데이터 변경 알림"""
        event = {
            'type': event_type,
            'collection': collection,
            'entity_id': entity_id,
            'data': data,
            'old_data': old_data,
            'timestamp': datetime.now().isoformat()
        }
        
        # 등록된 이벤트 핸들러들에게 알림
        for handler in self._event_handlers:
            await handler(event)
    
    def add_event_handler(self, handler):
        """이벤트 핸들러 등록"""
        self._event_handlers.append(handler)
    
    def get_metadata(self, collection: str = None) -> Dict[str, Any]:
        """메타데이터 조회"""
        if collection:
            return self._metadata.get(collection, {})
        return self._metadata

# 비즈니스 서비스 레이어 - 도메인 로직 구현
class CustomerService:
    """고객 서비스 - 비즈니스 로직 구현"""
    
    def __init__(self, customer_repo: CustomerRepository, order_repo: OrderRepository):
        self.customer_repo = customer_repo
        self.order_repo = order_repo
    
    async def create_customer(self, name: str, email: str) -> Customer:
        """고객 생성 - 비즈니스 규칙 적용"""
        # 중복 이메일 검사
        existing = await self.customer_repo.get_by_email(email)
        if existing:
            raise ValueError(f"Customer with email {email} already exists")
        
        customer = Customer(
            id=f"cust_{datetime.now().timestamp()}",
            name=name,
            email=email,
            created_at=datetime.now(),
            last_updated=datetime.now()
        )
        
        await self.customer_repo.create(customer)
        return customer
    
    async def get_customer_with_orders(self, customer_id: str) -> Dict[str, Any]:
        """고객과 주문 정보 통합 조회 - 데이터 조합"""
        customer = await self.customer_repo.get_by_id(customer_id)
        if not customer:
            return None
        
        orders = await self.order_repo.get_by_customer_id(customer_id)
        
        return {
            'customer': customer.to_dict(),
            'orders': [order.to_dict() for order in orders],
            'total_orders': len(orders),
            'total_amount': sum(order.amount for order in orders)
        }

# 이벤트 핸들러 - 데이터 변경 처리
async def audit_event_handler(event: Dict[str, Any]):
    """감사 로그 핸들러 - 데이터 변경 추적"""
    print(f"AUDIT: {event['type']} on {event['collection']}/{event['entity_id']} at {event['timestamp']}")

async def cache_invalidation_handler(event: Dict[str, Any]):
    """캐시 무효화 핸들러 - 데이터 일관성 보장"""
    print(f"CACHE: Invalidating cache for {event['collection']}/{event['entity_id']}")

# 데모 실행 함수
async def demo_data_centric_architecture():
    """데이터 중심 아키텍처 데모"""
    
    # 데이터 접근 계층 초기화
    dal = DataAccessLayer()
    
    # 이벤트 핸들러 등록
    dal.add_event_handler(audit_event_handler)
    dal.add_event_handler(cache_invalidation_handler)
    
    # 저장소 초기화
    customer_repo = CustomerRepository(dal)
    order_repo = OrderRepository(dal)
    
    # 서비스 초기화
    customer_service = CustomerService(customer_repo, order_repo)
    
    print("=== Data-Centric Architecture 데모 시작 ===\n")
    
    # 1. 고객 생성
    print("1. 고객 생성")
    customer = await customer_service.create_customer("홍길동", "hong@example.com")
    print(f"생성된 고객: {customer.name} ({customer.email})")
    
    # 2. 주문 생성
    print("\n2. 주문 생성")
    order1 = Order(
        id=f"order_1_{datetime.now().timestamp()}",
        customer_id=customer.id,
        amount=150000.0,
        status="completed",
        created_at=datetime.now()
    )
    await order_repo.create(order1)
    print(f"생성된 주문: {order1.id} - {order1.amount}원")
    
    order2 = Order(
        id=f"order_2_{datetime.now().timestamp()}",
        customer_id=customer.id,
        amount=75000.0,
        status="pending",
        created_at=datetime.now()
    )
    await order_repo.create(order2)
    print(f"생성된 주문: {order2.id} - {order2.amount}원")
    
    # 3. 통합 데이터 조회
    print("\n3. 고객과 주문 통합 조회")
    customer_data = await customer_service.get_customer_with_orders(customer.id)
    print(f"고객: {customer_data['customer']['name']}")
    print(f"총 주문 수: {customer_data['total_orders']}")
    print(f"총 주문 금액: {customer_data['total_amount']:,}원")
    
    # 4. 메타데이터 조회
    print("\n4. 메타데이터 정보")
    customers_meta = dal.get_metadata('customers')
    orders_meta = dal.get_metadata('orders')
    print(f"고객 컬렉션: {customers_meta['total_records']}개 레코드")
    print(f"주문 컬렉션: {orders_meta['total_records']}개 레코드")
    
    print("\n=== 데모 완료 ===")
    
    return {
        'customer': customer,
        'orders': [order1, order2],
        'metadata': dal.get_metadata()
    }

# 메인 실행
if __name__ == "__main__":
    # 비동기 실행
    result = asyncio.run(demo_data_centric_architecture())
    
    print(f"\n최종 결과:")
    print(f"생성된 고객 ID: {result['customer'].id}")
    print(f"생성된 주문 수: {len(result['orders'])}")
    print(f"전체 메타데이터: {json.dumps(result['metadata'], indent=2, ensure_ascii=False)}")

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

from kafka import KafkaConsumer
import json

# Kafka로부터 사용자 이벤트 수신
consumer = KafkaConsumer(
    'user-events',
    bootstrap_servers=['kafka-broker:9092'],
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for msg in consumer:
    event = msg.value
    if event['type'] == 'product_click':
        product_id = event['product_id']
        user_id = event['user_id']
        # 간단한 추천 로직 (캐시 예시)
        print(f"사용자 {user_id}가 클릭한 상품: {product_id} -> 추천 업데이트")