Batch Processing

  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

# 전자상거래 상품 추천 배치 처리 시스템
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import logging
import time

class RecommendationBatchProcessor:
    """
    상품 추천을 위한 배치 처리 시스템
    """
    
    def __init__(self, chunk_size=10000):
        """
        배치 프로세서 초기화
        
        Args:
            chunk_size: 청크 단위로 처리할 데이터 크기
        """
        self.chunk_size = chunk_size
        self.logger = self._setup_logger()
        
    def _setup_logger(self):
        """로깅 설정"""
        logging.basicConfig(level=logging.INFO)
        return logging.getLogger(__name__)
    
    def extract_user_behavior_data(self, start_date, end_date):
        """
        사용자 행동 데이터 추출 (ETL의 Extract 단계)
        
        Args:
            start_date: 데이터 추출 시작 날짜
            end_date: 데이터 추출 종료 날짜
            
        Returns:
            DataFrame: 사용자 행동 데이터
        """
        self.logger.info(f"사용자 행동 데이터 추출 시작: {start_date} ~ {end_date}")
        
        # 실제 구현에서는 데이터베이스에서 데이터를 추출
        # 여기서는 샘플 데이터 생성
        sample_data = {
            'user_id': np.random.randint(1, 10000, 50000),
            'product_id': np.random.randint(1, 1000, 50000),
            'action': np.random.choice(['view', 'cart', 'purchase'], 50000),
            'timestamp': pd.date_range(start_date, end_date, periods=50000)
        }
        
        return pd.DataFrame(sample_data)
    
    def transform_data(self, raw_data):
        """
        데이터 변환 및 전처리 (ETL의 Transform 단계)
        
        Args:
            raw_data: 원시 데이터
            
        Returns:
            DataFrame: 변환된 데이터
        """
        self.logger.info("데이터 변환 시작")
        
        # 사용자-상품 상호작용 매트릭스 생성
        user_item_matrix = raw_data.pivot_table(
            index='user_id', 
            columns='product_id', 
            values='action',
            aggfunc='count',
            fill_value=0
        )
        
        # 가중치 적용 (구매 > 장바구니 > 조회)
        weights = {'view': 1, 'cart': 3, 'purchase': 5}
        for action, weight in weights.items():
            mask = raw_data['action'] == action
            user_item_matrix += weight * mask.groupby(['user_id', 'product_id']).size().unstack(fill_value=0)
        
        return user_item_matrix
    
    def process_in_chunks(self, data, processing_func):
        """
        청크 단위로 데이터 처리
        
        Args:
            data: 처리할 데이터
            processing_func: 처리 함수
            
        Returns:
            결과 데이터
        """
        self.logger.info(f"청크 단위 처리 시작 (청크 크기: {self.chunk_size})")
        
        total_chunks = len(data) // self.chunk_size + 1
        results = []
        
        for i in range(0, len(data), self.chunk_size):
            chunk = data[i:i + self.chunk_size]
            chunk_result = processing_func(chunk)
            results.append(chunk_result)
            
            self.logger.info(f"청크 처리 완료: {i//self.chunk_size + 1}/{total_chunks}")
            
        return pd.concat(results) if results else pd.DataFrame()
    
    def generate_recommendations(self, user_item_matrix, top_k=10):
        """
        협업 필터링을 통한 추천 생성
        
        Args:
            user_item_matrix: 사용자-상품 매트릭스
            top_k: 추천할 상품 개수
            
        Returns:
            dict: 사용자별 추천 상품 리스트
        """
        self.logger.info("추천 생성 시작")
        
        # 코사인 유사도 계산
        user_similarity = cosine_similarity(user_item_matrix)
        
        recommendations = {}
        
        for user_idx, user_id in enumerate(user_item_matrix.index):
            # 유사한 사용자 찾기
            similar_users = np.argsort(user_similarity[user_idx])[::-1][1:11]  # 상위 10명
            
            # 추천 점수 계산
            scores = np.zeros(len(user_item_matrix.columns))
            
            for similar_user_idx in similar_users:
                similarity_score = user_similarity[user_idx][similar_user_idx]
                user_ratings = user_item_matrix.iloc[similar_user_idx].values
                scores += similarity_score * user_ratings
            
            # 이미 상호작용한 상품 제외
            user_interactions = user_item_matrix.iloc[user_idx].values
            scores[user_interactions > 0] = 0
            
            # 상위 K개 상품 추천
            top_products = np.argsort(scores)[::-1][:top_k]
            recommendations[user_id] = user_item_matrix.columns[top_products].tolist()
        
        return recommendations
    
    def load_recommendations(self, recommendations):
        """
        추천 결과 저장 (ETL의 Load 단계)
        
        Args:
            recommendations: 추천 결과
        """
        self.logger.info("추천 결과 저장 시작")
        
        # 실제 구현에서는 데이터베이스나 캐시에 저장
        # 여기서는 로깅으로 대체
        for user_id, products in recommendations.items():
            self.logger.info(f"사용자 {user_id}: {products}")
    
    def run_batch_job(self):
        """
        배치 작업 실행 메인 함수
        """
        start_time = time.time()
        self.logger.info("배치 작업 시작")
        
        try:
            # 1. 데이터 추출
            yesterday = datetime.now() - timedelta(days=1)
            today = datetime.now()
            
            raw_data = self.extract_user_behavior_data(yesterday, today)
            
            # 2. 데이터 변환
            user_item_matrix = self.transform_data(raw_data)
            
            # 3. 추천 생성 (청크 단위 처리)
            recommendations = self.generate_recommendations(user_item_matrix)
            
            # 4. 결과 저장
            self.load_recommendations(recommendations)
            
            # 5. 작업 완료 로깅
            end_time = time.time()
            processing_time = end_time - start_time
            self.logger.info(f"배치 작업 완료 (처리 시간: {processing_time:.2f}초)")
            
        except Exception as e:
            self.logger.error(f"배치 작업 실패: {str(e)}")
            raise

# 스케줄러와 함께 사용하는 예시
def daily_recommendation_job():
    """
    일일 추천 배치 작업
    """
    processor = RecommendationBatchProcessor(chunk_size=5000)
    processor.run_batch_job()

# 실행 예시
if __name__ == "__main__":
    daily_recommendation_job()

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

# EOD 집계 배치 예시 (PySpark)
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("EOD_Reconciliation").getOrCreate()

# 1. 데이터 추출
df = spark.read.jdbc(url=DB_URL, table="transactions",
                     properties=DB_PROPS).filter("date = current_date()")

# 2. 집계 변환
agg = df.groupBy("account_id").sum("amount").withColumnRenamed("sum(amount)", "daily_total")

# 3. 로드
agg.write.mode("overwrite").jdbc(url=DW_URL, table="eod_totals", properties=DW_PROPS)

# 4. 오류 대응 + 로깅
try:
    # 실행 완료
    pass
except Exception as e:
    # 알림, 재시도 로직 등
    raise
spark.stop()

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

# 예시: Python을 활용한 간단한 배치 처리 스크립트
import csv
import time

def batch_process(input_file, output_file):
    """
    대량의 데이터를 일괄적으로 처리하여 변환 결과를 저장하는 예시 함수
    """
    with open(input_file, 'r') as infile, open(output_file, 'w', newline='') as outfile:
        reader = csv.DictReader(infile)
        fieldnames = reader.fieldnames + ['processed_at']
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()
        for row in reader:
            # 데이터 변환 로직(예: 타임스탬프 추가)
            row['processed_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
            writer.writerow(row)

# 스케줄러(Cron 등)에서 정기적으로 호출
batch_process('input.csv', 'output.csv')