파이썬 Pandas로 다루는 대용량 데이터: 빅데이터 분석 완전 정복 가이드 2025 🐼 (2025)

🌟 Pandas의 주요 특징

1. 직관적인 데이터 구조: DataFrame과 Series라는 구조로 데이터를 쉽게 다룰 수 있어요.
2. 강력한 데이터 조작 기능: 필터링, 그룹화, 병합 등 다양한 데이터 처리가 가능해요.
3. 다양한 파일 형식 지원: CSV, Excel, SQL, JSON 등 거의 모든 데이터 형식을 읽고 쓸 수 있어요.
4. 빠른 성능: C로 작성된 내부 코드 덕분에 대용량 데이터도 빠르게 처리해요.
5. 데이터 시각화 연동: Matplotlib, Seaborn 등과 쉽게 연동되어 데이터 시각화가 간편해요.

🔍 빅데이터의 3V

1. Volume (양): 테라바이트, 페타바이트 단위의 엄청난 양의 데이터
2. Velocity (속도): 데이터가 생성되고 처리되는 빠른 속도
3. Variety (다양성): 구조화된 데이터부터 비구조화된 데이터까지 다양한 형태

2025년에는 여기에 Veracity (정확성)와 Value (가치)가 추가되어 5V로 확장되었어요!

🚧 빅데이터 분석의 주요 도전 과제

1. 메모리 제한: 일반 PC의 RAM으로는 모든 데이터를 한 번에 로드하기 어려워요.
2. 처리 시간: 대용량 데이터 처리는 시간이 오래 걸릴 수 있어요.
3. 데이터 품질: 대용량 데이터에는 오류, 중복, 누락된 값이 많을 수 있어요.
4. 복잡한 분석: 데이터가 커질수록 패턴 발견이 더 복잡해질 수 있어요.
5. 시각화 어려움: 너무 많은 데이터를 어떻게 의미 있게 시각화할 것인가?

1. 청크 단위 처리: 데이터를 작은 조각으로 나눠서 순차적으로 처리해요.
2. 메모리 최적화: 데이터 타입 최적화로 메모리 사용량을 줄여요.
3. 분산 처리 연동: Dask, Spark 같은 분산 처리 프레임워크와 연동해요.
4. SQL 쿼리 활용: 데이터베이스에서 필요한 부분만 가져와 처리해요.
5. 파일 포맷 최적화: Parquet, HDF5 같은 효율적인 파일 형식을 활용해요.

📦 Pandas 설치하기

pip install pandas

최신 버전(2025년 3월 기준 v2.2.x)을 설치하려면:

pip install pandas==2.2.0

🧰 빅데이터 분석을 위한 추가 라이브러리

1. NumPy: 수치 계산의 기본이 되는 라이브러리
2. Matplotlib & Seaborn: 데이터 시각화 라이브러리
3. Dask: 대용량 데이터의 병렬 처리를 위한 라이브러리
4. PyArrow: Apache Arrow를 Python에서 사용하기 위한 라이브러리
5. Vaex: 메모리보다 큰 표 형식 데이터를 처리하는 라이브러리

🔹 Series: 1차원 배열

Series는 라벨이 있는 1차원 배열이에요. 엑셀의 한 열(column)이라고 생각하시면 쉬워요.

# Series 생성하기import pandas as pd# 리스트로부터 Series 만들기s = pd.Series([1, 3, 5, 7, 9])print(s)# 인덱스 지정하기s = pd.Series([1, 3, 5, 7, 9], index=['a', 'b', 'c', 'd', 'e'])print(s)# 딕셔너리로부터 Series 만들기data = {'a': 1, 'b': 3, 'c': 5, 'd': 7, 'e': 9}s = pd.Series(data)print(s)

🔹 DataFrame: 2차원 테이블

DataFrame은 여러 개의 Series가 모인 2차원 테이블이에요. 엑셀 시트와 비슷하다고 생각하시면 됩니다.

# DataFrame 생성하기import pandas as pd# 딕셔너리로부터 DataFrame 만들기data = { '이름': ['김데이터', '이분석', '박파이썬', '최판다스'], '나이': [25, 30, 28, 32], '직업': ['데이터 사이언티스트', '백엔드 개발자', '프론트엔드 개발자', '데이터 엔지니어'], '연봉': [5000, 4500, 4000, 5500]}df = pd.DataFrame(data)print(df)# CSV 파일에서 DataFrame 불러오기# df = pd.read_csv('data.csv')# Excel 파일에서 DataFrame 불러오기# df = pd.read_excel('data.xlsx')

🔍 DataFrame 기본 조작

# 기본 정보 확인하기print(df.shape) # 행과 열의 개수print(df.columns) # 열 이름들print(df.dtypes) # 각 열의 데이터 타입print(df.head()) # 처음 5행 보기print(df.tail()) # 마지막 5행 보기print(df.info()) # 데이터프레임 정보 요약print(df.describe()) # 수치형 열의 통계 요약# 데이터 접근하기print(df['이름']) # 열 접근print(df.loc[0]) # 행 접근 (라벨 기반)print(df.iloc[0]) # 행 접근 (위치 기반)print(df.loc[0, '이름']) # 특정 값 접근# 데이터 필터링print(df[df['나이'] > 28]) # 나이가 28보다 큰 행들print(df[(df['나이'] > 28) & (df['연봉'] > 5000)]) # 조건 조합# 데이터 정렬print(df.sort_values('연봉', ascending=False)) # 연봉 기준 내림차순 정렬# 새 열 추가하기df['연봉_등급'] = ['A' if x > 5000 else 'B' if x > 4500 else 'C' for x in df['연봉']]print(df)

🔄 대용량 CSV 파일 효율적으로 불러오기

# 청크 단위로 CSV 파일 읽기import pandas as pd# 한 번에 10,000행씩 처리chunk_size = 10000chunks = []for chunk in pd.read_csv('big_data.csv', chunksize=chunk_size): # 각 청크마다 필요한 처리 수행 processed_chunk = chunk[chunk['중요도'] > 3] # 예: 필터링 chunks.append(processed_chunk) # 처리된 청크들 합치기result = pd.concat(chunks, ignore_index=True)print(f"처리된 데이터 크기: {result.shape}")

또는 특정 열만 선택해서 메모리 사용량을 줄일 수도 있어요:

# 필요한 열만 선택해서 읽기selected_columns = ['날짜', '제품명', '판매량', '가격']df = pd.read_csv('big_data.csv', usecols=selected_columns)print(f"선택된 데이터 크기: {df.shape}")

📂 효율적인 파일 형식

1. Parquet 파일 사용하기

Parquet은 컬럼 기반 저장 방식으로, 압축률이 높고 읽기 속도가 빠른 파일 형식이에요.

# Parquet 파일로 저장하기df.to_parquet('data.parquet', compression='snappy')# Parquet 파일 읽기df = pd.read_parquet('data.parquet')# 특정 열만 읽기df = pd.read_parquet('data.parquet', columns=['날짜', '판매량'])

2. HDF5 파일 사용하기

HDF5는 계층적 데이터 형식으로, 대용량 데이터를 효율적으로 저장하고 접근할 수 있어요.

# HDF5 파일로 저장하기df.to_hdf('data.h5', key='df', mode='w')# HDF5 파일 읽기df = pd.read_hdf('data.h5', key='df')# 여러 데이터프레임 저장하기df1.to_hdf('data.h5', key='df1', mode='w')df2.to_hdf('data.h5', key='df2', mode='a') # append 모드

3. Feather 파일 사용하기

Feather는 빠른 읽기/쓰기 속도를 제공하는 파일 형식이에요.

# Feather 파일로 저장하기df.to_feather('data.feather')# Feather 파일 읽기df = pd.read_feather('data.feather')

☁️ 클라우드 스토리지에서 직접 데이터 불러오기

# AWS S3에서 Parquet 파일 읽기import s3fsfs = s3fs.S3FileSystem(anon=False) # 인증 필요df = pd.read_parquet('s3://my-bucket/data.parquet', filesystem=fs)# Google Cloud Storage에서 CSV 파일 읽기import gcsfsfs = gcsfs.GCSFileSystem(project='my-project')with fs.open('gs://my-bucket/big_data.csv') as f: df = pd.read_csv(f)

🔍 데이터 타입 최적화

데이터 타입을 적절히 선택하면 메모리 사용량을 크게 줄일 수 있어요.

# 데이터프레임의 메모리 사용량 확인def memory_usage(df): return f"메모리 사용량: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB"# 원본 데이터프레임df = pd.read_csv('big_data.csv')print(memory_usage(df))# 데이터 타입 최적화def optimize_dtypes(df): # 정수형 열 최적화 for col in df.select_dtypes(include=['int']).columns: col_min = df[col].min() col_max = df[col].max() # 적절한 정수 타입 선택 if col_min >= 0: if col_max < 2**8: df[col] = df[col].astype('uint8') elif col_max < 2**16: df[col] = df[col].astype('uint16') elif col_max < 2**32: df[col] = df[col].astype('uint32') else: if col_min > -2**7 and col_max < 2**7: df[col] = df[col].astype('int8') elif col_min > -2**15 and col_max < 2**15: df[col] = df[col].astype('int16') elif col_min > -2**31 and col_max < 2**31: df[col] = df[col].astype('int32') # 부동소수점 열 최적화 for col in df.select_dtypes(include=['float']).columns: df[col] = df[col].astype('float32') # 범주형 데이터 최적화 for col in df.select_dtypes(include=['object']).columns: if df[col].nunique() < df.shape[0] * 0.5: # 고유값이 적으면 범주형으로 변환 df[col] = df[col].astype('category') return df# 최적화 적용df_optimized = optimize_dtypes(df)print(memory_usage(df_optimized))print(f"메모리 절약: {(1 - df_optimized.memory_usage(deep=True).sum() / df.memory_usage(deep=True).sum()) * 100:.2f}%")

🧩 청크 처리 기법

전체 데이터를 한 번에 처리하지 않고, 작은 조각(청크)으로 나눠서 처리하면 메모리 사용량을 크게 줄일 수 있어요.

# 청크 단위로 처리하고 결과 집계하기def process_in_chunks(filename, chunk_size=10000): # 예: 각 청크에서 평균 계산 후 전체 평균 구하기 total_sum = 0 total_count = 0 for chunk in pd.read_csv(filename, chunksize=chunk_size): # 필요한 열만 선택 chunk = chunk[['판매량', '가격']] # 각 청크에서 계산 수행 chunk['매출'] = chunk['판매량'] * chunk['가격'] # 결과 집계 total_sum += chunk['매출'].sum() total_count += len(chunk) # 메모리에서 청크 삭제 (명시적으로 가비지 컬렉션 호출) del chunk import gc gc.collect() # 최종 결과 계산 average_sales = total_sum / total_count return average_sales# 청크 단위로 처리avg_sales = process_in_chunks('sales_data.csv', chunk_size=50000)print(f"평균 매출: {avg_sales}")

⏳ 지연 평가(Lazy Evaluation) 활용하기

# 2025년 Pandas의 지연 평가 기능 활용import pandas as pd# 지연 평가 모드 활성화 (2025년 신기능)pd.options.mode.lazy_evaluation = True# 대용량 데이터 불러오기df = pd.read_parquet('huge_data.parquet')# 여러 변환 작업 정의 (아직 실행되지 않음)df = df[df['중요도'] > 3]df = df.groupby('카테고리').agg({'판매량': 'sum', '가격': 'mean'})df = df.sort_values('판매량', ascending=False)# 실제로 필요한 시점에 계산 실행top_categories = df.head(10) # 이 시점에서 모든 계산이 실행됨print(top_categories)

🚀 GPU 가속 활용하기

# RAPIDS cuDF를 활용한 GPU 가속 (2025년 버전)import cudf# GPU로 Parquet 파일 불러오기gdf = cudf.read_parquet('huge_data.parquet')# GPU에서 데이터 처리 (CPU보다 훨씬 빠름)result = gdf.groupby('카테고리').agg({'판매량': 'sum', '가격': 'mean'})# 결과를 다시 CPU로 가져오기cpu_result = result.to_pandas()print(cpu_result)

🧼 결측치(Missing Values) 처리하기

대용량 데이터에서는 결측치를 효율적으로 처리하는 것이 중요해요.

# 결측치 확인하기missing_values = df.isnull().sum()print(missing_values)# 결측치 비율 확인하기missing_percentage = (df.isnull().sum() / len(df)) * 100print(missing_percentage)# 결측치가 많은 열 삭제하기 (예: 50% 이상이 결측치인 경우)threshold = 50.0drop_cols = missing_percentage[missing_percentage > threshold].indexdf_cleaned = df.drop(columns=drop_cols)print(f"삭제된 열: {list(drop_cols)}")# 결측치 채우기 - 수치형 데이터df_cleaned['판매량'].fillna(df_cleaned['판매량'].mean(), inplace=True) # 평균으로 채우기df_cleaned['가격'].fillna(df_cleaned['가격'].median(), inplace=True) # 중앙값으로 채우기# 결측치 채우기 - 범주형 데이터df_cleaned['카테고리'].fillna(df_cleaned['카테고리'].mode()[0], inplace=True) # 최빈값으로 채우기# 고급 결측치 처리: KNN 기반 결측치 대체 (대용량 데이터에 최적화)from sklearn.impute import KNNImputer# 메모리 효율을 위해 샘플링 후 KNN 적용sample_size = min(100000, len(df_cleaned))df_sample = df_cleaned.sample(sample_size, random_state=42)# 수치형 열에만 KNN 적용numeric_cols = df_sample.select_dtypes(include=['float', 'int']).columnsimputer = KNNImputer(n_neighbors=5)df_sample[numeric_cols] = imputer.fit_transform(df_sample[numeric_cols])# 학습된 imputer를 전체 데이터에 청크 단위로 적용chunk_size = 50000for i in range(0, len(df_cleaned), chunk_size): end = min(i + chunk_size, len(df_cleaned)) df_cleaned.iloc[i:end, df_cleaned.columns.get_indexer(numeric_cols)] = imputer.transform( df_cleaned.iloc[i:end][numeric_cols] )

🔄 데이터 변환 및 정규화

대용량 데이터에서는 메모리 효율적인 방식으로 데이터 변환을 수행해야 해요.

# 범주형 데이터 인코딩 (메모리 효율적 방식)for col in df.select_dtypes(include=['object', 'category']).columns: # 고유값이 많지 않은 경우에만 원-핫 인코딩 적용 if df[col].nunique() < 10: # get_dummies는 메모리를 많이 사용할 수 있으므로 주의 dummies = pd.get_dummies(df[col], prefix=col, drop_first=True) df = pd.concat([df.drop(columns=[col]), dummies], axis=1) else: # 고유값이 많은 경우 라벨 인코딩 적용 from sklearn.preprocessing import LabelEncoder le = LabelEncoder() df[col] = le.fit_transform(df[col].astype(str))# 수치형 데이터 정규화 (청크 단위로 처리)from sklearn.preprocessing import StandardScaler# 스케일러 학습 (샘플 데이터로)numeric_cols = df.select_dtypes(include=['float', 'int']).columnssample_size = min(100000, len(df))scaler = StandardScaler()scaler.fit(df.sample(sample_size)[numeric_cols])# 청크 단위로 스케일링 적용chunk_size = 50000for i in range(0, len(df), chunk_size): end = min(i + chunk_size, len(df)) df.iloc[i:end, df.columns.get_indexer(numeric_cols)] = scaler.transform( df.iloc[i:end][numeric_cols] )

🔍 이상치(Outlier) 탐지 및 처리

대용량 데이터에서 이상치를 효율적으로 탐지하고 처리하는 방법이에요.

# 통계적 방법으로 이상치 탐지 (IQR 방법)def detect_outliers(df, column): Q1 = df[column].quantile(0.25) Q3 = df[column].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)] return outliers, lower_bound, upper_bound# 청크 단위로 이상치 처리하기def process_outliers_in_chunks(filename, column, chunk_size=10000): # 먼저 전체 데이터의 통계치 계산 (샘플링 사용) sample_chunks = [] for i, chunk in enumerate(pd.read_csv(filename, chunksize=chunk_size)): if i < 10: # 처음 10개 청크만 사용 sample_chunks.append(chunk[column]) sample_data = pd.concat(sample_chunks) Q1 = sample_data.quantile(0.25) Q3 = sample_data.quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR # 이제 전체 데이터를 청크 단위로 처리 result_chunks = [] outlier_count = 0 for chunk in pd.read_csv(filename, chunksize=chunk_size): # 이상치 처리 (예: 경계값으로 대체) outliers_mask = (chunk[column] < lower_bound) | (chunk[column] > upper_bound) outlier_count += outliers_mask.sum() # 이상치를 경계값으로 대체 chunk.loc[chunk[column] < lower_bound, column] = lower_bound chunk.loc[chunk[column] > upper_bound, column] = upper_bound result_chunks.append(chunk) # 처리된 데이터 합치기 result_df = pd.concat(result_chunks, ignore_index=True) print(f"처리된 이상치 개수: {outlier_count}") return result_df# 함수 사용 예cleaned_df = process_outliers_in_chunks('sales_data.csv', '판매량', chunk_size=50000)

🤖 자동화된 데이터 전처리 (2025년 기술)

# AutoML 기반 데이터 전처리 (2025년 버전)from auto_preprocessor import DataCleaner # 가상의 2025년 라이브러리# 메모리 효율적인 자동 전처리기 초기화cleaner = DataCleaner(memory_efficient=True, n_jobs=-1)# 청크 단위로 데이터 전처리processed_chunks = []for chunk in pd.read_csv('big_data.csv', chunksize=50000): # 자동으로 결측치 처리, 이상치 탐지, 인코딩, 정규화 등을 수행 processed_chunk = cleaner.transform(chunk) processed_chunks.append(processed_chunk)# 처리된 데이터 합치기processed_df = pd.concat(processed_chunks, ignore_index=True)print(f"전처리 완료된 데이터 크기: {processed_df.shape}")

📊 대용량 데이터 시각화 기본 전략

1. 샘플링(Sampling): 전체 데이터의 일부만 추출하여 시각화
2. 집계(Aggregation): 데이터를 그룹화하고 요약 통계를 시각화
3. 빈닝(Binning): 연속적인 데이터를 구간으로 나누어 시각화
4. 점진적 렌더링: 데이터를 점진적으로 로드하며 시각화
5. 인터랙티브 시각화: 사용자가 관심 있는 부분을 확대/축소할 수 있는 기능 제공

🔍 샘플링을 통한 시각화

# 대용량 데이터에서 샘플링하여 산점도 그리기import matplotlib.pyplot as pltimport seaborn as sns# 전체 데이터에서 10,000개 샘플 추출sample_size = 10000df_sample = df.sample(sample_size, random_state=42)# 산점도 그리기plt.figure(figsize=(10, 6))sns.scatterplot(data=df_sample, x='특성1', y='특성2', hue='카테고리', alpha=0.6)plt.title('대용량 데이터의 샘플 산점도')plt.show()# 층화 샘플링으로 균형 잡힌 시각화from sklearn.model_selection import StratifiedShuffleSplit# 카테고리별로 균등하게 샘플링splitter = StratifiedShuffleSplit(n_splits=1, test_size=sample_size/len(df), random_state=42)for _, sample_idx in splitter.split(df, df['카테고리']): df_stratified = df.iloc[sample_idx]plt.figure(figsize=(10, 6))sns.scatterplot(data=df_stratified, x='특성1', y='특성2', hue='카테고리', alpha=0.6)plt.title('층화 샘플링을 통한 균형 잡힌 산점도')plt.show()

📊 집계를 통한 시각화

# 대용량 데이터를 집계하여 시각화# 예: 시간별 판매량 추이# 청크 단위로 데이터를 읽고 집계하기time_sales = {}for chunk in pd.read_csv('sales_data.csv', chunksize=50000): # 날짜 열을 datetime으로 변환 chunk['날짜'] = pd.to_datetime(chunk['날짜']) # 시간별로 집계 daily_sales = chunk.groupby(chunk['날짜'].dt.date)['판매량'].sum() # 기존 집계 결과와 병합 for date, sales in daily_sales.items(): if date in time_sales: time_sales[date] += sales else: time_sales[date] = sales# 집계 결과를 데이터프레임으로 변환sales_df = pd.DataFrame(list(time_sales.items()), columns=['날짜', '판매량'])sales_df = sales_df.sort_values('날짜')# 시계열 그래프 그리기plt.figure(figsize=(12, 6))plt.plot(sales_df['날짜'], sales_df['판매량'])plt.title('일별 총 판매량 추이')plt.xlabel('날짜')plt.ylabel('판매량')plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()# 히트맵으로 시각화 (요일별, 시간별 패턴)for chunk in pd.read_csv('sales_data.csv', chunksize=50000): chunk['날짜'] = pd.to_datetime(chunk['날짜']) chunk['요일'] = chunk['날짜'].dt.day_name() chunk['시간'] = chunk['날짜'].dt.hour # 요일 및 시간별 집계 hourly_sales = chunk.groupby(['요일', '시간'])['판매량'].sum().unstack() # 첫 번째 청크만 처리 (예시) breakplt.figure(figsize=(12, 8))sns.heatmap(hourly_sales, cmap='YlGnBu', annot=True, fmt='.0f')plt.title('요일 및 시간별 판매량 히트맵')plt.tight_layout()plt.show()

📊 빈닝을 통한 시각화

# 연속적인 데이터를 구간으로 나누어 시각화# 예: 가격 분포 히스토그램# 청크 단위로 데이터를 읽고 빈(bin) 카운트 계산import numpy as np# 빈의 경계 정의 (사전 지식 또는 첫 번째 청크로부터 결정)price_bins = np.linspace(0, 1000, 51) # 0-1000 사이를 50개 구간으로 나눔bin_counts = np.zeros(len(price_bins) - 1)for chunk in pd.read_csv('sales_data.csv', chunksize=50000): # 각 청크에서 히스토그램 계산 chunk_counts, _ = np.histogram(chunk['가격'], bins=price_bins) # 전체 카운트에 추가 bin_counts += chunk_counts# 히스토그램 그리기plt.figure(figsize=(12, 6))plt.bar(price_bins[:-1], bin_counts, width=price_bins[1] - price_bins[0], alpha=0.7, align='edge')plt.title('제품 가격 분포')plt.xlabel('가격')plt.ylabel('제품 수')plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()# 2D 히스토그램 (가격 vs 판매량)# 첫 번째 청크로 빈 경계 결정first_chunk = next(pd.read_csv('sales_data.csv', chunksize=1000))price_bins = np.linspace(first_chunk['가격'].min(), first_chunk['가격'].max(), 30)sales_bins = np.linspace(first_chunk['판매량'].min(), first_chunk['판매량'].max(), 30)# 2D 히스토그램 초기화hist_2d = np.zeros((len(price_bins) - 1, len(sales_bins) - 1))for chunk in pd.read_csv('sales_data.csv', chunksize=50000): # 각 청크에서 2D 히스토그램 계산 chunk_hist, _, _ = np.histogram2d( chunk['가격'], chunk['판매량'], bins=[price_bins, sales_bins] ) # 전체 히스토그램에 추가 hist_2d += chunk_hist# 2D 히스토그램 그리기plt.figure(figsize=(10, 8))plt.pcolormesh(price_bins, sales_bins, hist_2d.T, cmap='viridis')plt.colorbar(label='제품 수')plt.title('가격 vs 판매량 분포')plt.xlabel('가격')plt.ylabel('판매량')plt.tight_layout()plt.show()

🔍 인터랙티브 시각화 (2025년 기술)

# Plotly와 Dash를 활용한 인터랙티브 대시보드import plotly.express as pximport dashfrom dash import dcc, htmlfrom dash.dependencies import Input, Output# 데이터 준비 (집계된 데이터 사용)# 실제로는 대용량 데이터를 미리 집계하여 사용# 대시보드 앱 생성app = dash.Dash(__name__)app.layout = html.Div([ html.H1("대용량 판매 데이터 분석 대시보드"), html.Div([ html.Label("카테고리 선택:"), dcc.Dropdown( id='category-dropdown', options=[{'label': cat, 'value': cat} for cat in sales_df['카테고리'].unique()], value='전체', multi=True ), html.Label("날짜 범위:"), dcc.DatePickerRange( id='date-range', start_date=sales_df['날짜'].min(), end_date=sales_df['날짜'].max() ) ]), dcc.Graph(id='time-series-chart'), html.Div([ dcc.Graph(id='category-pie-chart', style={'display': 'inline-block', 'width': '50%'}), dcc.Graph(id='price-histogram', style={'display': 'inline-block', 'width': '50%'}) ])])# 콜백 함수 정의@app.callback( [Output('time-series-chart', 'figure'), Output('category-pie-chart', 'figure'), Output('price-histogram', 'figure')], [Input('category-dropdown', 'value'), Input('date-range', 'start_date'), Input('date-range', 'end_date')])def update_charts(categories, start_date, end_date): # 필터링된 데이터 가져오기 filtered_df = sales_df if categories and '전체' not in categories: filtered_df = filtered_df[filtered_df['카테고리'].isin(categories)] filtered_df = filtered_df[ (filtered_df['날짜'] >= start_date) & (filtered_df['날짜'] <= end_date) ] # 시계열 차트 time_fig = px.line( filtered_df, x='날짜', y='판매량', title='기간별 판매량 추이' ) # 파이 차트 pie_data = filtered_df.groupby('카테고리')['판매량'].sum().reset_index() pie_fig = px.pie( pie_data, values='판매량', names='카테고리', title='카테고리별 판매 비중' ) # 히스토그램 hist_fig = px.histogram( filtered_df, x='가격', title='가격 분포', nbins=50 ) return time_fig, pie_fig, hist_fig# 앱 실행if __name__ == '__main__': app.run_server(debug=True)

🚀 GPU 가속 시각화 (2025년 기술)

# datashader를 활용한 대용량 데이터 시각화import datashader as dsimport datashader.transfer_functions as tffrom datashader.colors import Viridisimport colorcet as cc# 캔버스 생성canvas = ds.Canvas(plot_width=800, plot_height=600)# 대용량 데이터 시각화 (파케이 파일에서 직접)df = pd.read_parquet('huge_data.parquet')# 포인트 렌더링agg = canvas.points(df, 'x', 'y')img = tf.shade(agg, cmap=cc.fire)img.to_pil().save('point_viz.png')# 히트맵 렌더링agg = canvas.heatmap(df, 'x', 'y', aggregator=ds.count())img = tf.shade(agg, cmap=Viridis)img.to_pil().save('heatmap_viz.png')# 라인 렌더링 (시계열 데이터)agg = canvas.line(df, 'time', 'value', agg=ds.count())img = tf.shade(agg, cmap=cc.blues)img.to_pil().save('line_viz.png')

🔄 Dask와 Pandas 연동

Dask는 Pandas와 유사한 API를 제공하면서도 분산 처리를 지원하는 라이브러리예요.

# Dask와 Pandas 연동하기import pandas as pdimport dask.dataframe as dd# Pandas DataFrame을 Dask DataFrame으로 변환pandas_df = pd.read_csv('sample.csv')dask_df = dd.from_pandas(pandas_df, npartitions=4)# 또는 대용량 CSV 파일을 직접 Dask로 읽기dask_df = dd.read_csv('big_data_*.csv') # 여러 파일 한 번에 읽기 가능# Dask로 대용량 데이터 처리result = dask_df[dask_df['value'] > 100].groupby('category').mean().compute()# 결과를 다시 Pandas로 변환pandas_result = result.compute()# Dask의 지연 계산 활용# 계산 그래프 정의 (아직 실행되지 않음)future_result = (dask_df .map_partitions(lambda df: df[df['value'] > 100]) .groupby('category') .mean())# 필요한 시점에 계산 실행final_result = future_result.compute()

🔄 Apache Spark와 Pandas 연동

Spark는 대규모 분산 데이터 처리에 최적화된 프레임워크예요.

# PySpark와 Pandas 연동하기import pandas as pdfrom pyspark.sql import SparkSession# Spark 세션 생성spark = SparkSession.builder \ .appName("PandasIntegration") \ .config("spark.executor.memory", "8g") \ .getOrCreate()# Pandas DataFrame을 Spark DataFrame으로 변환pandas_df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})spark_df = spark.createDataFrame(pandas_df)# 또는 대용량 데이터를 직접 Spark로 읽기spark_df = spark.read.csv('hdfs://big_data.csv', header=True, inferSchema=True)# Spark로 대용량 데이터 처리result_spark = spark_df.filter(spark_df.value > 100).groupBy("category").mean()# 결과를 다시 Pandas로 변환result_pandas = result_spark.toPandas()# Pandas UDFs (사용자 정의 함수) 활용from pyspark.sql.functions import pandas_udffrom pyspark.sql.types import DoubleType# Pandas UDF 정의@pandas_udf(DoubleType())def complex_transformation(s: pd.Series) -> pd.Series: # 복잡한 Pandas 연산 수행 return s.fillna(s.mean()).apply(lambda x: x**2 + 1)# Spark DataFrame에 Pandas UDF 적용spark_df = spark_df.withColumn("transformed", complex_transformation(spark_df.value))

🔄 Vaex와 Pandas 연동

Vaex는 메모리보다 큰 표 형식 데이터를 처리하는 데 최적화된 라이브러리예요.

# Vaex와 Pandas 연동하기import pandas as pdimport vaex# Pandas DataFrame을 Vaex DataFrame으로 변환pandas_df = pd.read_csv('sample.csv')vaex_df = vaex.from_pandas(pandas_df)# 또는 대용량 데이터를 직접 Vaex로 읽기vaex_df = vaex.open('big_data.hdf5') # HDF5, CSV, Parquet 등 지원# Vaex로 대용량 데이터 처리 (메모리 효율적)filtered = vaex_df[vaex_df.value > 100]result = filtered.groupby('category').agg({'value': 'mean'})# 결과를 다시 Pandas로 변환pandas_result = result.to_pandas_df()# Vaex의 지연 계산 및 메모리 매핑 활용# 대용량 CSV를 메모리 효율적으로 처리vaex_df = vaex.from_csv('huge_data.csv', convert=True, chunk_size=5_000_000)# 복잡한 계산 수행 (메모리 제한 없이)result = vaex_df.evaluate(vaex_df.value.mean())print(f"평균값: {result}")

☁️ 클라우드 빅데이터 서비스와 Pandas 연동

# AWS Athena와 Pandas 연동import pandas as pdimport awswrangler as wr# Athena에서 SQL 쿼리 실행 후 결과를 Pandas로 가져오기query = "SELECT * FROM my_database.my_table WHERE value > 100 LIMIT 1000"df = wr.athena.read_sql_query(query, database="my_database")# 결과를 S3에 저장wr.s3.to_parquet( df=df, path="s3://my-bucket/my-folder/", dataset=True, partition_cols=["category"])# Google BigQuery와 Pandas 연동from google.cloud import bigquery# BigQuery 클라이언트 생성client = bigquery.Client()# SQL 쿼리 실행 후 결과를 Pandas로 가져오기query = """ SELECT * FROM `my-project.my_dataset.my_table` WHERE value > 100 LIMIT 1000"""df = client.query(query).to_dataframe()# 결과를 BigQuery에 저장df.to_gbq( destination_table="my_dataset.results", project_id="my-project", if_exists="replace")

🔄 실시간 데이터 스트리밍과 Pandas 연동

# Kafka와 Pandas 연동import pandas as pdfrom confluent_kafka import Consumerimport json# Kafka 컨슈머 설정conf = { 'bootstrap.servers': 'localhost:9092', 'group.id': 'pandas-group', 'auto.offset.reset': 'earliest'}consumer = Consumer(conf)consumer.subscribe(['my-topic'])# 스트리밍 데이터를 Pandas로 처리def process_stream(batch_size=1000, timeout=1.0): messages = [] while len(messages) < batch_size: msg = consumer.poll(timeout) if msg is None: break if msg.error(): print(f"Consumer error: {msg.error()}") continue # 메시지 파싱 try: value = json.loads(msg.value().decode('utf-8')) messages.append(value) except Exception as e: print(f"Error parsing message: {e}") # 배치 데이터를 Pandas DataFrame으로 변환 if messages: df = pd.DataFrame(messages) # 데이터 처리 result = df.groupby('category').agg({'value': ['mean', 'count']}) print(result) # 처리 결과 저장 또는 다른 시스템으로 전송 # ... return len(messages)# 스트리밍 처리 루프while True: processed = process_stream() if processed == 0: print("No more messages, waiting...") import time time.sleep(5)

📊 프로젝트 시나리오: 대규모 이커머스 판매 데이터 분석

데이터 규모: 3년간의 판매 데이터, 약 5억 건의 거래 기록 (약 200GB)

목표:

1. 판매 트렌드 분석 및 예측
2. 고객 세그먼트 분석
3. 제품 추천 시스템 개발
4. 재고 최적화 전략 수립

도전 과제: 일반 PC에서 이 대용량 데이터를 효율적으로 처리하고 분석해야 함

🛠️ 1단계: 데이터 로딩 및 전처리

# 1. 효율적인 파일 형식으로 변환import pandas as pdimport dask.dataframe as ddimport os# 원본 CSV 파일을 Parquet로 변환 (한 번만 수행)def convert_to_parquet(csv_dir, parquet_dir): # 모든 CSV 파일 목록 가져오기 csv_files = [os.path.join(csv_dir, f) for f in os.listdir(csv_dir) if f.endswith('.csv')] for csv_file in csv_files: # 파일명에서 날짜 추출 (예: sales_2023_01.csv -> 2023_01) date_part = os.path.basename(csv_file).split('_')[1:3] date_str = '_'.join(date_part) # 청크 단위로 읽고 Parquet으로 변환 for i, chunk in enumerate(pd.read_csv(csv_file, chunksize=1000000)): # 데이터 타입 최적화 for col in chunk.select_dtypes(include=['int']).columns: chunk[col] = pd.to_numeric(chunk[col], downcast='integer') for col in chunk.select_dtypes(include=['float']).columns: chunk[col] = pd.to_numeric(chunk[col], downcast='float') # 날짜 열 변환 chunk['order_date'] = pd.to_datetime(chunk['order_date']) # Parquet 파일로 저장 (날짜별로 디렉토리 구성) output_dir = os.path.join(parquet_dir, date_str) os.makedirs(output_dir, exist_ok=True) chunk.to_parquet( os.path.join(output_dir, f'part_{i:05d}.parquet'), engine='pyarrow', compression='snappy' ) print(f"Converted {csv_file} to Parquet")# 변환 실행convert_to_parquet('raw_data', 'parquet_data')# 2. Dask를 사용하여 전체 데이터셋 로드ddf = dd.read_parquet('parquet_data/*/*.parquet')# 3. 기본 정보 확인print(f"데이터 크기: {len(ddf)} 행, {len(ddf.columns)} 열")print(f"컬럼 목록: {list(ddf.columns)}")print(ddf.dtypes)# 4. 데이터 전처리# 결측치 확인missing_counts = ddf.isnull().sum().compute()print(f"결측치 개수:\n{missing_counts}")# 중복 제거ddf = ddf.drop_duplicates(subset=['order_id', 'product_id'])# 이상치 처리 (예: 음수 가격 수정)ddf = ddf.map_partitions( lambda df: df.assign(price=df['price'].clip(lower=0)))# 파생 변수 생성ddf['year'] = ddf['order_date'].dt.yearddf['month'] = ddf['order_date'].dt.monthddf['day'] = ddf['order_date'].dt.dayddf['day_of_week'] = ddf['order_date'].dt.dayofweekddf['total_amount'] = ddf['price'] * ddf['quantity']# 5. 전처리된 데이터 저장ddf.to_parquet( 'processed_data/sales.parquet', engine='pyarrow', compression='snappy', write_index=False)

🔍 2단계: 판매 트렌드 분석

# 1. 시간별 판매 트렌드 분석import pandas as pdimport dask.dataframe as ddimport matplotlib.pyplot as pltimport seaborn as sns# 전처리된 데이터 로드ddf = dd.read_parquet('processed_data/sales.parquet')# 일별 판매량 및 매출 집계daily_sales = ddf.groupby(ddf['order_date'].dt.date).agg({ 'order_id': 'nunique', # 주문 수 'quantity': 'sum', # 판매 수량 'total_amount': 'sum' # 총 매출}).compute()# 시계열 그래프 그리기plt.figure(figsize=(15, 10))# 일별 매출 추이plt.subplot(3, 1, 1)plt.plot(daily_sales.index, daily_sales['total_amount'], 'b-')plt.title('일별 총 매출')plt.grid(True, alpha=0.3)# 일별 주문 수 추이plt.subplot(3, 1, 2)plt.plot(daily_sales.index, daily_sales['order_id'], 'g-')plt.title('일별 주문 수')plt.grid(True, alpha=0.3)# 일별 판매 수량 추이plt.subplot(3, 1, 3)plt.plot(daily_sales.index, daily_sales['quantity'], 'r-')plt.title('일별 판매 수량')plt.grid(True, alpha=0.3)plt.tight_layout()plt.savefig('daily_sales_trends.png', dpi=300)plt.close()# 2. 월별, 요일별 패턴 분석# 월별 매출 집계monthly_sales = ddf.groupby(['year', 'month']).agg({ 'total_amount': 'sum'}).compute().reset_index()# 월별 매출 히트맵monthly_pivot = monthly_sales.pivot(index='year', columns='month', values='total_amount')plt.figure(figsize=(12, 8))sns.heatmap(monthly_pivot, annot=True, fmt='.1f', cmap='YlGnBu')plt.title('월별 매출 히트맵')plt.savefig('monthly_sales_heatmap.png', dpi=300)plt.close()# 요일별 매출 패턴dow_sales = ddf.groupby('day_of_week').agg({ 'total_amount': 'sum'}).compute()dow_sales.index = ['월', '화', '수', '목', '금', '토', '일']plt.figure(figsize=(10, 6))sns.barplot(x=dow_sales.index, y=dow_sales['total_amount'])plt.title('요일별 총 매출')plt.savefig('dow_sales.png', dpi=300)plt.close()# 3. 카테고리별 판매 트렌드category_sales = ddf.groupby(['year', 'month', 'category']).agg({ 'total_amount': 'sum'}).compute().reset_index()# 상위 5개 카테고리 추출top_categories = ddf.groupby('category').agg({ 'total_amount': 'sum'}).compute().nlargest(5, 'total_amount').index# 상위 카테고리의 월별 매출 트렌드plt.figure(figsize=(15, 8))for category in top_categories: cat_data = category_sales[category_sales['category'] == category] cat_data['date'] = pd.to_datetime(cat_data['year'].astype(str) + '-' + cat_data['month'].astype(str)) plt.plot(cat_data['date'], cat_data['total_amount'], label=category)plt.title('상위 5개 카테고리의 월별 매출 트렌드')plt.legend()plt.grid(True, alpha=0.3)plt.savefig('category_trends.png', dpi=300)plt.close()

👥 3단계: 고객 세그먼트 분석

# 1. RFM 분석 (Recency, Frequency, Monetary)import pandas as pdimport dask.dataframe as ddimport numpy as npimport matplotlib.pyplot as pltimport seaborn as snsfrom datetime import datetime# 전처리된 데이터 로드ddf = dd.read_parquet('processed_data/sales.parquet')# 분석 기준일 설정 (데이터셋의 마지막 날짜)max_date = ddf['order_date'].max().compute()# 고객별 RFM 지표 계산rfm = ddf.groupby('customer_id').agg({ 'order_date': 'max', # 최근 구매일 (Recency) 'order_id': 'nunique', # 구매 빈도 (Frequency) 'total_amount': 'sum' # 총 구매금액 (Monetary)}).compute()# Recency 계산 (일 단위)rfm['recency'] = (max_date - rfm['order_date']).dt.daysrfm.rename(columns={'order_id': 'frequency', 'total_amount': 'monetary'}, inplace=True)# RFM 점수 계산 (4분위수 기준)quantiles = rfm[['recency', 'frequency', 'monetary']].quantile([0.25, 0.5, 0.75])rfm_score = rfm.copy()# Recency 점수 (낮을수록 좋음)rfm_score['r_score'] = np.where(rfm['recency'] <= quantiles.loc[0.25, 'recency'], 4, np.where(rfm['recency'] <= quantiles.loc[0.5, 'recency'], 3, np.where(rfm['recency'] <= quantiles.loc[0.75, 'recency'], 2, 1)))# Frequency 점수 (높을수록 좋음)rfm_score['f_score'] = np.where(rfm['frequency'] >= quantiles.loc[0.75, 'frequency'], 4, np.where(rfm['frequency'] >= quantiles.loc[0.5, 'frequency'], 3, np.where(rfm['frequency'] >= quantiles.loc[0.25, 'frequency'], 2, 1)))# Monetary 점수 (높을수록 좋음)rfm_score['m_score'] = np.where(rfm['monetary'] >= quantiles.loc[0.75, 'monetary'], 4, np.where(rfm['monetary'] >= quantiles.loc[0.5, 'monetary'], 3, np.where(rfm['monetary'] >= quantiles.loc[0.25, 'monetary'], 2, 1)))# 종합 RFM 점수rfm_score['rfm_score'] = rfm_score['r_score'] + rfm_score['f_score'] + rfm_score['m_score']# 고객 세그먼트 정의segment_map = { 12: '최우수 고객', 11: '우수 고객', 10: '우수 고객', 9: '우수 고객', 8: '잠재 충성 고객', 7: '잠재 충성 고객', 6: '일반 고객', 5: '일반 고객', 4: '관심 필요 고객', 3: '이탈 위험 고객'}# 9점 이하는 모두 '이탈 위험 고객'으로 처리rfm_score['segment'] = rfm_score['rfm_score'].apply(lambda x: segment_map.get(x, '이탈 위험 고객'))# 세그먼트별 고객 수 시각화plt.figure(figsize=(12, 6))segment_counts = rfm_score['segment'].value_counts().sort_values(ascending=False)sns.barplot(x=segment_counts.index, y=segment_counts.values)plt.title('고객 세그먼트 분포')plt.xticks(rotation=45)plt.tight_layout()plt.savefig('customer_segments.png', dpi=300)plt.close()# 세그먼트별 평균 RFM 값 분석segment_rfm = rfm_score.groupby('segment').agg({ 'recency': 'mean', 'frequency': 'mean', 'monetary': 'mean', 'customer_id': 'count' # 고객 수}).rename(columns={'customer_id': 'count'})print(segment_rfm)# 세그먼트별 특성 시각화 (레이더 차트)from math import pi# 데이터 준비categories = ['Recency', 'Frequency', 'Monetary']N = len(categories)# 값 정규화 (0-1 사이로)segment_rfm_norm = segment_rfm.copy()for col in ['recency', 'frequency', 'monetary']: min_val = segment_rfm[col].min() max_val = segment_rfm[col].max() segment_rfm_norm[col] = (segment_rfm[col] - min_val) / (max_val - min_val) # Recency는 낮을수록 좋으므로 반전segment_rfm_norm['recency'] = 1 - segment_rfm_norm['recency']# 상위 4개 세그먼트만 시각화top_segments = ['최우수 고객', '우수 고객', '잠재 충성 고객', '일반 고객']# 레이더 차트 그리기plt.figure(figsize=(10, 10))ax = plt.subplot(111, polar=True)# 각도 설정angles = [n / float(N) * 2 * pi for n in range(N)]angles += angles[:1] # 원형으로 닫기 위해 첫 각도 추가# 축 그리기plt.xticks(angles[:-1], categories)ax.set_rlabel_position(0)plt.yticks([0.2, 0.4, 0.6, 0.8], ["0.2", "0.4", "0.6", "0.8"], color="grey", size=7)plt.ylim(0, 1)# 각 세그먼트 그리기for i, segment in enumerate(top_segments): values = segment_rfm_norm.loc[segment, ['recency', 'frequency', 'monetary']].values.tolist() values += values[:1] # 원형으로 닫기 위해 첫 값 추가 ax.plot(angles, values, linewidth=2, linestyle='solid', label=segment) ax.fill(angles, values, alpha=0.1)plt.legend(loc='upper right', bbox_to_anchor=(0.1, 0.1))plt.title('고객 세그먼트별 RFM 특성')plt.tight_layout()plt.savefig('segment_radar.png', dpi=300)plt.close()

🔮 4단계: 제품 추천 시스템 개발

# 1. 협업 필터링 기반 추천 시스템import pandas as pdimport numpy as npfrom scipy.sparse import csr_matrixfrom sklearn.neighbors import NearestNeighborsimport implicitfrom tqdm import tqdm# 구매 데이터 로드 (메모리 제한으로 인해 최근 1년 데이터만 사용)recent_date = pd.Timestamp('2024-03-16')one_year_ago = recent_date - pd.Timedelta(days=365)# 청크 단위로 데이터 로드 및 필터링purchase_data = []for chunk in pd.read_parquet('processed_data/sales.parquet', engine='pyarrow'): # 최근 1년 데이터만 필터링 chunk = chunk[chunk['order_date'] >= one_year_ago] purchase_data.append(chunk[['customer_id', 'product_id', 'quantity']])purchase_df = pd.concat(purchase_data)print(f"최근 1년 구매 데이터: {len(purchase_df)} 행")# 고객-제품 매트릭스 생성purchase_matrix = purchase_df.pivot_table( index='customer_id', columns='product_id', values='quantity', aggfunc='sum', fill_value=0)# 희소 행렬로 변환 (메모리 효율성)sparse_purchase = csr_matrix(purchase_matrix.values)# 모델 학습 (Item-based Collaborative Filtering)model = implicit.als.AlternatingLeastSquares( factors=50, regularization=0.1, iterations=20)# 학습 (구매 수량을 신뢰도로 사용)model.fit(sparse_purchase.T) # 아이템 기반이므로 전치행렬 사용# 추천 함수def recommend_products(customer_idx, model, purchase_matrix, n_recommendations=5): # 이미 구매한 제품 필터링 already_purchased = purchase_matrix.iloc[customer_idx].to_numpy().nonzero()[0] # 추천 받기 recommendations = model.recommend( customer_idx, sparse_purchase[customer_idx], N=n_recommendations + len(already_purchased), filter_already_liked_items=True ) # 제품 ID와 점수 반환 product_ids = [purchase_matrix.columns[idx] for idx, _ in recommendations] scores = [score for _, score in recommendations] return product_ids[:n_recommendations], scores[:n_recommendations]# 샘플 고객에 대한 추천sample_customers = np.random.choice(purchase_matrix.shape[0], 5, replace=False)for idx in sample_customers: customer_id = purchase_matrix.index[idx] recommended_products, scores = recommend_products(idx, model, purchase_matrix) print(f"고객 ID {customer_id}에 대한 추천 제품:") for product, score in zip(recommended_products, scores): print(f" - 제품 ID: {product}, 점수: {score:.4f}") print()# 2. 연관 규칙 분석 (장바구니 분석)from mlxtend.frequent_patterns import apriori, association_rules# 트랜잭션 데이터 준비 (주문별 제품 목록)transactions = purchase_df.groupby('order_id')['product_id'].apply(list).reset_index()# 원-핫 인코딩 (메모리 효율을 위해 상위 1000개 제품만 사용)top_products = purchase_df['product_id'].value_counts().nlargest(1000).index# 청크 단위로 처리encoded_chunks = []chunk_size = 10000for i in range(0, len(transactions), chunk_size): chunk = transactions.iloc[i:i+chunk_size] # 원-핫 인코딩 encoded_chunk = pd.DataFrame(index=chunk['order_id']) for product in top_products: encoded_chunk[product] = chunk['product_id'].apply(lambda x: 1 if product in x else 0) encoded_chunks.append(encoded_chunk)encoded_df = pd.concat(encoded_chunks)# 빈발 아이템 집합 찾기frequent_itemsets = apriori(encoded_df, min_support=0.01, use_colnames=True)# 연관 규칙 생성rules = association_rules(frequent_itemsets, metric="lift", min_threshold=1.0)rules = rules.sort_values('lift', ascending=False)# 상위 10개 연관 규칙 출력print("상위 10개 제품 연관 규칙:")for i, row in rules.head(10).iterrows(): antecedents = ', '.join([str(x) for x in row['antecedents']]) consequents = ', '.join([str(x) for x in row['consequents']]) print(f"{antecedents} → {consequents} (지지도: {row['support']:.4f}, 신뢰도: {row['confidence']:.4f}, 향상도: {row['lift']:.4f})")

📦 5단계: 재고 최적화 전략

# 1. 제품별 판매 예측 및 재고 최적화import pandas as pdimport numpy as npfrom prophet import Prophetimport matplotlib.pyplot as pltfrom tqdm import tqdm# 일별 제품 판매량 데이터 준비# 메모리 효율을 위해 상위 판매 제품만 분석sales_data = []for chunk in pd.read_parquet('processed_data/sales.parquet', engine='pyarrow'): # 일별, 제품별 판매량 집계 daily_sales = chunk.groupby(['order_date', 'product_id'])['quantity'].sum().reset_index() sales_data.append(daily_sales)sales_df = pd.concat(sales_data)# 상위 100개 판매 제품 선택top_products = sales_df.groupby('product_id')['quantity'].sum().nlargest(100).index# 제품별 시계열 예측forecasts = {}for product_id in tqdm(top_products, desc="제품별 판매 예측"): # 제품별 판매 데이터 추출 product_sales = sales_df[sales_df['product_id'] == product_id] product_sales = product_sales.groupby('order_date')['quantity'].sum().reset_index() # Prophet 형식으로 변환 prophet_df = product_sales.rename(columns={'order_date': 'ds', 'quantity': 'y'}) # 결측일 채우기 (판매 없는 날은 0으로) date_range = pd.date_range(prophet_df['ds'].min(), prophet_df['ds'].max()) prophet_df = prophet_df.set_index('ds').reindex(date_range, fill_value=0).reset_index() prophet_df = prophet_df.rename(columns={'index': 'ds'}) # 모델 학습 model = Prophet( yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=False, seasonality_mode='multiplicative' ) model.fit(prophet_df) # 향후 30일 예측 future = model.make_future_dataframe(periods=30) forecast = model.predict(future) # 예측 결과 저장 forecasts[product_id] = forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']] # 시각화 (상위 10개 제품만) if product_id in top_products[:10]: fig = model.plot(forecast) plt.title(f'제품 {product_id} 판매량 예측') plt.savefig(f'forecast_product_{product_id}.png', dpi=300) plt.close()# 2. 재고 최적화 전략 수립# 안전 재고 수준 계산 (서비스 수준 95% 가정)from scipy import statssafety_stock = {}lead_times = {} # 제품별 리드타임 (실제로는 제품 데이터에서 가져와야 함)# 임의의 리드타임 할당 (실제로는 제품 데이터에서 가져와야 함)for product_id in top_products: lead_times[product_id] = np.random.randint(1, 10) # 1~10일 사이 리드타임for product_id in top_products: forecast = forecasts[product_id] # 예측 기간(30일) 동안의 일별 수요 daily_demand = forecast['yhat'].tail(30).values # 수요의 표준편차 demand_std = np.std(daily_demand) # 리드타임 lead_time = lead_times[product_id] # 서비스 수준 95%에 해당하는 Z값 z_score = stats.norm.ppf(0.95) # 안전 재고 = Z * 수요 표준편차 * sqrt(리드타임) safety_stock[product_id] = z_score * demand_std * np.sqrt(lead_time)# 경제적 주문량(EOQ) 계산# EOQ = sqrt(2 * 연간 수요 * 주문 비용 / 보관 비용)order_cost = 100 # 주문 비용 (고정)holding_cost_rate = 0.2 # 보관 비용 비율 (제품 가치의 20%)eoq = {}reorder_point = {}for product_id in top_products: forecast = forecasts[product_id] # 일별 예측 수요의 평균 daily_demand_avg = forecast['yhat'].tail(30).mean() # 연간 수요 추정 annual_demand = daily_demand_avg * 365 # 제품 가격 (실제로는 제품 데이터에서 가져와야 함) # 여기서는 임의의 값 사용 product_price = np.random.uniform(10, 100) # 보관 비용 holding_cost = product_price * holding_cost_rate # 경제적 주문량 계산 eoq[product_id] = np.sqrt(2 * annual_demand * order_cost / holding_cost) # 재주문 시점 = 리드타임 동안의 평균 수요 + 안전 재고 reorder_point[product_id] = daily_demand_avg * lead_times[product_id] + safety_stock[product_id]# 결과 출력inventory_df = pd.DataFrame({ 'product_id': list(top_products), 'avg_daily_demand': [forecasts[p]['yhat'].tail(30).mean() for p in top_products], 'safety_stock': [safety_stock[p] for p in top_products], 'lead_time': [lead_times[p] for p in top_products], 'eoq': [eoq[p] for p in top_products], 'reorder_point': [reorder_point[p] for p in top_products]})print("재고 최적화 결과 (상위 10개 제품):")print(inventory_df.head(10))

🧠 메모리 사용 최적화 팁

1. 불필요한 열 제거하기: 필요한 열만 로드하면 메모리 사용량을 크게 줄일 수 있어요.

# 필요한 열만 선택df = pd.read_csv('big_data.csv', usecols=['date', 'product_id', 'quantity', 'price'])

2. 데이터 타입 최적화하기: 적절한 데이터 타입을 사용하면 메모리 사용량을 크게 줄일 수 있어요.

# 정수형 열 최적화df['quantity'] = pd.to_numeric(df['quantity'], downcast='integer')# 부동소수점 열 최적화df['price'] = pd.to_numeric(df['price'], downcast='float')# 범주형 데이터 변환df['category'] = df['category'].astype('category')

3. 불필요한 인덱스 제거하기: 기본 정수 인덱스를 사용하면 메모리를 절약할 수 있어요.

# 인덱스 재설정df.reset_index(drop=True, inplace=True)

4. 객체 복사 최소화하기: 불필요한 복사본을 만들지 않도록 주의해요.

# 나쁜 예 (복사본 생성)df_filtered = df[df['value'] > 0]df_transformed = df_filtered.copy()df_transformed['value'] = df_transformed['value'] * 2# 좋은 예 (연쇄 연산)df_result = df[df['value'] > 0].assign(value=lambda x: x['value'] * 2)

5. 메모리 회수 명시적으로 하기: 대용량 객체를 사용한 후에는 명시적으로 메모리를 회수해요.

# 대용량 객체 사용 후 메모리 회수del large_dfimport gcgc.collect()

⚡ 처리 속도 향상 팁

1. 벡터화 연산 사용하기: 루프 대신 벡터화된 연산을 사용하면 훨씬 빠르게 처리할 수 있어요.

# 나쁜 예 (루프 사용)for i in range(len(df)): df.loc[i, 'result'] = df.loc[i, 'value'] * 2# 좋은 예 (벡터화 연산)df['result'] = df['value'] * 2

2. apply() 대신 벡터화 연산 사용하기: 가능하면 apply() 대신 벡터화된 연산을 사용해요.

# 나쁜 예 (apply 사용)df['result'] = df['text'].apply(lambda x: x.upper())# 좋은 예 (벡터화 연산)df['result'] = df['text'].str.upper()

3. groupby 연산 최적화하기: 필요한 열만 그룹화하면 처리 속도가 빨라져요.

# 나쁜 예 (전체 DataFrame 그룹화)result = df.groupby('category').mean()# 좋은 예 (필요한 열만 그룹화)result = df.groupby('category')['value'].mean()

4. query() 메서드 활용하기: 복잡한 필터링에는 query() 메서드가 효율적이에요.

# 나쁜 예 (여러 조건 조합)filtered = df[(df['value'] > 100) & (df['category'] == 'A') | (df['value'] < 0)]# 좋은 예 (query 사용)filtered = df.query("(value > 100 and category == 'A') or (value < 0)")

5. SQL 쿼리 활용하기: 복잡한 연산은 때로는 SQL이 더 효율적일 수 있어요.

# pandasql 사용from pandasql import sqldfquery = """SELECT category, AVG(value) as avg_valueFROM dfWHERE value > 0GROUP BY categoryHAVING COUNT(*) > 10ORDER BY avg_value DESC"""result = sqldf(query, locals())

⚙️ 병렬 처리 활용 팁

1. Pandas 내장 병렬 처리 활성화하기: 2025년 Pandas는 내장 병렬 처리 기능이 강화되었어요.

# Pandas 병렬 처리 설정 (2025년 버전)pd.options.compute.use_parallel = Truepd.options.compute.parallel_backend = 'processes' # 'threads' 또는 'processes'pd.options.compute.parallel_workers = 8 # CPU 코어 수에 맞게 설정

2. Swifter 라이브러리 활용하기: apply 연산을 자동으로 병렬화해주는 라이브러리예요.

# swifter 설치: pip install swifterimport swifter# 자동으로 최적의 방법 선택df['result'] = df['text'].swifter.apply(lambda x: complex_function(x))

3. Dask 활용하기: 대용량 데이터는 Dask를 활용해 병렬 처리해요.

import dask.dataframe as dd# Dask DataFrame으로 변환ddf = dd.from_pandas(df, npartitions=8) # CPU 코어 수에 맞게 설정# 병렬 처리 수행result = ddf.map_partitions(lambda part: complex_function(part)).compute()

4. Modin 활용하기: Pandas API를 그대로 사용하면서 병렬 처리하는 라이브러리예요.

# modin 설치: pip install modin[ray]import modin.pandas as pd# 기존 Pandas 코드와 동일하게 사용하되 자동으로 병렬 처리됨df = pd.read_csv('big_data.csv')result = df.groupby('category').mean()

💾 캐싱 및 중간 결과 저장 팁

1. 중간 결과 저장하기: 시간이 오래 걸리는 연산의 결과는 저장해두면 좋아요.

# 중간 결과 Parquet으로 저장intermediate_result = df.groupby('category').agg({'value': ['mean', 'sum', 'count']})intermediate_result.to_parquet('intermediate_result.parquet')# 나중에 다시 불러오기intermediate_result = pd.read_parquet('intermediate_result.parquet')

2. 메모이제이션 활용하기: 동일한 연산을 반복할 때는 결과를 캐싱해요.

from functools import lru_cache@lru_cache(maxsize=128)def complex_calculation(category): # 시간이 오래 걸리는 계산 return df[df['category'] == category]['value'].mean()# 결과가 캐싱됨result1 = complex_calculation('A')result2 = complex_calculation('A') # 캐시에서 바로 반환

3. Joblib 활용하기: 대용량 객체의 결과를 디스크에 캐싱해요.

from joblib import Memory# 캐시 디렉토리 설정memory = Memory('cache_dir', verbose=0)@memory.cachedef process_large_data(filename): # 시간이 오래 걸리는 데이터 처리 df = pd.read_csv(filename) result = complex_processing(df) return result# 처음 호출 시 계산 후 캐싱result = process_large_data('big_data.csv')# 두 번째 호출 시 캐시에서 로드result = process_large_data('big_data.csv')

📊 성능 모니터링 팁

1. 메모리 사용량 모니터링하기: 대용량 데이터 처리 시 메모리 사용량을 지속적으로 확인해요.

import psutilimport osdef check_memory_usage(): process = psutil.Process(os.getpid()) memory_info = process.memory_info() memory_usage_mb = memory_info.rss / 1024 / 1024 return f"현재 메모리 사용량: {memory_usage_mb:.2f} MB"# 주요 연산 전후로 메모리 사용량 확인print(check_memory_usage())result = df.groupby('category').agg({'value': ['mean', 'sum']})print(check_memory_usage())

2. 코드 실행 시간 측정하기: 성능 병목을 찾기 위해 실행 시간을 측정해요.

import timedef timeit(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} 실행 시간: {end_time - start_time:.4f}초") return result return wrapper@timeitdef process_data(df): # 데이터 처리 코드 return df.groupby('category').mean()result = process_data(df)

3. line_profiler 활용하기: 코드 라인별 실행 시간을 분석해요.

# line_profiler 설치: pip install line_profiler# 프로파일링할 함수에 데코레이터 추가@profiledef process_data(df): result1 = df.groupby('category').mean() # 이 라인의 실행 시간 측정 result2 = df.query("value > 0") # 이 라인의 실행 시간 측정 result3 = pd.merge(result1, result2, on='category') # 이 라인의 실행 시간 측정 return result3# 명령줄에서 실행: kernprof -l script.py# 결과 확인: python -m line_profiler script.py.lprof

🛠️ 핵심 기술 스택 로드맵

1. 기초 단계

1. 프로그래밍 기초: Python 기본 문법, 자료구조, 함수, 클래스
2. 데이터 분석 라이브러리: Pandas, NumPy 기본 사용법
3. 데이터 시각화: Matplotlib, Seaborn 기초
4. SQL 기초: 기본 쿼리, 조인, 집계 함수
5. 통계 기초: 기술 통계, 확률 분포, 가설 검정

2. 중급 단계

1. 고급 Pandas: 대용량 데이터 처리, 성능 최적화
2. 데이터 전처리: 결측치 처리, 이상치 탐지, 특성 공학
3. 고급 시각화: Plotly, Bokeh, 대시보드 구축
4. 머신러닝 기초: scikit-learn, 지도학습, 비지도학습
5. 빅데이터 도구: Dask, Spark 기초

3. 고급 단계

1. 분산 컴퓨팅: Spark, Hadoop 생태계
2. 클라우드 서비스: AWS, GCP, Azure 데이터 서비스
3. 고급 머신러닝: 앙상블, 하이퍼파라미터 튜닝
4. 딥러닝: TensorFlow, PyTorch
5. MLOps: 모델 배포, 모니터링, 파이프라인 구축

4. 전문가 단계 (2025년 트렌드)

1. AI 기반 데이터 분석: AutoML, 생성형 AI 활용
2. 실시간 데이터 처리: 스트리밍 데이터 분석, 실시간 대시보드
3. 데이터 거버넌스: 데이터 품질, 보안, 규정 준수
4. 특화 분야 전문성: 금융, 의료, 마케팅 등 도메인 지식
5. 데이터 제품 개발: 데이터 기반 서비스 및 제품 설계

📚 추천 학습 자원

1. 온라인 강좌

1. Coursera: "Data Science with Python" 시리즈
2. edX: MIT의 "Data Analysis for Social Scientists"
3. Udemy: "Python for Data Science and Machine Learning Bootcamp"
4. DataCamp: "Data Scientist with Python" 트랙
5. 재능넷: 다양한 데이터 분석 강의와 멘토링 서비스

2. 책

1. "Python for Data Analysis" by Wes McKinney (Pandas 창시자)
2. "Hands-On Machine Learning with Scikit-Learn, Keras, and TensorFlow" by Aurélien Géron
3. "Big Data Processing with Apache Spark" by Bogdan Ghit
4. "Data Science from Scratch" by Joel Grus
5. "Storytelling with Data" by Cole Nussbaumer Knaflic

3. 프로젝트 및 실습

1. Kaggle: 데이터 과학 경진대회 및 데이터셋
2. GitHub: 오픈 소스 프로젝트 기여
3. 공공 데이터: 공공 데이터 포털의 데이터셋 활용
4. 개인 프로젝트: 관심 분야의 데이터 수집 및 분석
5. 재능넷 프로젝트: 실제 클라이언트의 데이터 분석 프로젝트 참여

💼 빅데이터 분야 직업 경로

1. 데이터 분석가

데이터를 분석하여 비즈니스 인사이트를 도출하는 역할

• 필요 기술: SQL, Pandas, 시각화, 통계, 비즈니스 이해력
• 연봉 범위 (2025년 기준): 5,000만원 ~ 8,000만원

2. 데이터 엔지니어

데이터 파이프라인을 구축하고 대용량 데이터 처리 시스템을 관리하는 역할

• 필요 기술: SQL, Spark, Hadoop, 클라우드 서비스, 데이터베이스
• 연봉 범위 (2025년 기준): 6,000만원 ~ 9,000만원

3. 데이터 사이언티스트

고급 분석 및 머신러닝 모델을 개발하여 복잡한 문제를 해결하는 역할

• 필요 기술: Python, R, 통계, 머신러닝, 딥러닝, 수학
• 연봉 범위 (2025년 기준): 7,000만원 ~ 1억 2,000만원

4. 머신러닝 엔지니어

머신러닝 모델을 프로덕션 환경에 배포하고 유지보수하는 역할

• 필요 기술: Python, 머신러닝, MLOps, 소프트웨어 엔지니어링
• 연봉 범위 (2025년 기준): 8,000만원 ~ 1억 3,000만원

5. 데이터 아키텍트

조직의 데이터 전략을 수립하고 데이터 인프라를 설계하는 역할

• 필요 기술: 데이터 모델링, 클라우드 아키텍처, 데이터 거버넌스
• 연봉 범위 (2025년 기준): 9,000만원 ~ 1억 5,000만원

🌟 성공을 위한 팁

1. 포트폴리오 구축: GitHub에 프로젝트를 공유하고 블로그에 분석 결과를 게시하세요.
2. 네트워킹: 데이터 커뮤니티에 참여하고 컨퍼런스에 참석하세요.
3. 도메인 지식 습득: 특정 산업 분야에 대한 전문성을 키우세요.
4. 소통 능력 개발: 기술적 내용을 비기술자에게 설명할 수 있는 능력을 키우세요.
5. 지속적 학습: 빠르게 변화하는 기술 트렌드를 따라가세요.
6. 실무 경험 쌓기: 재능넷과 같은 플랫폼에서 실제 프로젝트에 참여하세요.
7. 인증 및 자격증: 관련 자격증을 취득하여 전문성을 인정받으세요.