저는 최근 암호화폐 시장 데이터 파이프라인을 구축하면서 예기치 못한 문제에 직면했습니다. Tardis API에서 수신하는 BTC/USDT Tick 데이터가 하루 만에 2.3GB에 달했고, 월간 비용이 $340를 초과하기 시작한 것이었죠. 압축 없이 원본 CSV를 계속 사용하다간 월간 스토리지 비용만 $800을 넘길 판이었습니다.
다행스럽게도 Parquet 포맷 전환과 컬럼 기반 압축을 통해 2.3GB → 420MB로 줄이는 데 성공했습니다. 이 글에서는 실제 구현 과정과踩坑 경험까지 상세히 공유하겠습니다.
왜 Tardis CSV에서 Parquet로 전환해야 하는가
Tardis Machine API는 Binance, Bybit, OKX 등 24개 거래소의 실시간 Tick 데이터를 제공합니다. 그러나 CSV 포맷에는 치명적 단점이 있습니다:
- 반복 데이터 과다: 각 행에 심볼, 거래소 이름이 문자열로 반복 저장
- 타입 정보 부재: 모든 데이터가 텍스트로 저장되어 파싱 오버헤드 발생
- 압축 미지원: gzip/zstd 같은 인코딩 옵션 없음
- 쿼리 성능: 전체 파일 스캔 필수, 시간 범위 필터링 불가
실제 구현: Tardis CSV → Parquet 변환 파이프라인
제가 구축한 파이프라인은 Python + Apache Arrow 기반으로, 실제 환경에서 검증된 코드입니다.
# tardis_to_parquet.py
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime, timedelta
import os
class TardisToParquetConverter:
def __init__(self, output_dir: str = "./data/parquet"):
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def load_csv_chunk(self, csv_path: str, chunk_size: int = 100_000):
"""대용량 CSV를 청크 단위로 읽기"""
for chunk in pd.read_csv(
csv_path,
chunksize=chunk_size,
dtype={
'symbol': 'str',
'exchange': 'str',
'price': 'float32', # float64 → float32 (63% 메모리 절감)
'volume': 'float32',
'timestamp': 'int64' # 문자열 → Unix ms 정수
}
):
yield chunk
def optimize_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
"""Parquet 최적화를 위한 데이터 타입 변환"""
# 1. 타임스탬프를 datetime으로 변환 후 Unix 밀리초로 저장
df['timestamp'] = pd.to_datetime(df['timestamp']).astype('int64')
# 2. 카테고리 인코딩 (반복 문자열 압축)
for col in ['symbol', 'exchange', 'side']:
df[col] = df[col].astype('category')
# 3. 불필요한 컬럼 제거
drop_cols = [c for c in ['local_timestamp', 'sequence'] if c in df.columns]
df = df.drop(columns=drop_cols, errors='ignore')
return df
def convert_file(self, csv_path: str, partition_by: str = 'date'):
"""단일 CSV 파일을 Parquet로 변환"""
table_list = []
for chunk in self.load_csv_chunk(csv_path):
optimized = self.optimize_dataframe(chunk)
table_list.append(
pa.Table.from_pandas(optimized, preserve_index=False)
)
# 청크 병합
combined = pa.concat_tables(table_list)
# 파티션 분할 저장 (날짜별)
if partition_by == 'date':
dates = pd.to_datetime(combined.column('timestamp').to_pylist(), unit='ms')
for date in dates.unique():
mask = dates == date
partition_table = combined.filter(
pa.compute.equal(dates.cast(pa.int64()), int(date.value))
)
output_path = f"{self.output_dir}/date={date.strftime('%Y-%m-%d')}/data.parquet"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
pq.write_table(
partition_table,
output_path,
compression='zstd', # zstd: gzip 대비 15% 추가 압축
compression_level=3,
use_dictionary=True # 문자열 컬럼 딕셔너리 인코딩
)
return len(combined)
사용 예시
converter = TardisToParquetConverter(output_dir="./btc_tick_data")
row_count = converter.convert_file("./raw/tardis_btcusdt_2024.csv")
print(f"변환 완료: {row_count:,} 행")
# batch_convert.py - 배치 처리 스크립트
import glob
import os
from concurrent.futures import ThreadPoolExecutor
import time
def convert_directory(input_pattern: str, max_workers: int = 4):
"""여러 CSV 파일 병렬 변환"""
converter = TardisToParquetConverter()
csv_files = glob.glob(input_pattern)
total_start = time.time()
success_count = 0
def process_file(csv_path):
try:
filename = os.path.basename(csv_path)
print(f"변환 중: {filename}")
rows = converter.convert_file(csv_path)
return (filename, rows, None)
except Exception as e:
return (csv_path, 0, str(e))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(process_file, csv_files))
total_time = time.time() - total_start
for filename, rows, error in results:
if error:
print(f"❌ {filename}: {error}")
else:
print(f"✅ {filename}: {rows:,} 행 ({total_time/len(results):.1f}s/file)")
return results
실행
convert_directory("./raw/**/*.csv")
예상 성능: 8코어 머신에서 시간당 ~50GB CSV 처리 가능
if __name__ == "__main__":
convert_directory("./raw/*.csv", max_workers=4)
저장 공간 비교: CSV vs Parquet vs Parquet+Zstd
실제 30일치 BTC/USDT Tardis 데이터(총 1억 2천만 행)로 테스트한 결과입니다: