금융 데이터를 다루는 백엔드 개발자 분들이라면 한 번쯤 마주쳤을 오류입니다. Tardis.dev에서 수백만 건의 트레이딩 데이터를 CSV로 내보냈다가 서버 용량이 순식간에 바닥나고, 파싱 시간만 30분이 넘게 걸리는 경험을 해본 적 없으신가요? 오늘은 이 고통스러운 데이터 처리 과정을 Parquet 포맷으로 최적화하여 저장 공간을 90% 이상 절감하고 처리 속도를 비약적으로 높이는 방법을 실제 코드와 함께 설명드리겠습니다.

Tardis.dev란?

HolySheep AI 플랫폼의 백엔드 엔지니어로 일하며 다양한 금융 데이터 소스를 다루고 있습니다. Tardis.dev는 암호화폐 및 전통 금융市场的 실시간 및 이력 데이터를 제공하는 플랫폼으로, WebSocket 스트리밍과 REST API 두 가지 방식으로 데이터를 받을 수 있습니다. 하지만 원본 JSON 형식 그대로 저장하면 용량이 과도하게 커지고, Parquet로 변환하면 분석 속도와 스토리지 효율성이 극대화됩니다.

왜 Parquet인가?

Parquet는 Apache Hadoop 생태계에서 탄생한 열 기반(columnar) 스토리지 포맷입니다. CSV 대비 압축률이 5~10배 높고, 필요한 열만 읽어오는 기능 덕분에 분석 쿼리 속도가 드라마틱하게 향상됩니다. 특히 Tardis.dev에서 받는 고빈도 트레이딩 데이터(밀리초 단위의 틱 데이터)를 Parquet로 변환하면:

실전 프로젝트 구조

# 프로젝트 디렉토리 구조
tardis-parquet-pipeline/
├── src/
│   ├── config.py           # 설정 및 API 키 관리
│   ├── connector.py        # Tardis.dev API 연결
│   ├── transformer.py      # JSON → Parquet 변환 로직
│   ├── storage.py          # S3/MinIO 스토리지 연동
│   └── main.py             # 메인 파이프라인 실행
├── tests/
│   └── test_transformer.py # 단위 테스트
├── requirements.txt
└── .env.example

1단계: 환경 설정 및 의존성 설치

# requirements.txt
pandas>=2.0.0
pyarrow>=14.0.0
fastparquet>=2023.10.0
pyTardisDevClient>=1.0.0
python-dotenv>=1.0.0
boto3>=1.28.0
asyncio>=3.4.3


설치 명령어

pip install pandas pyarrow fastparquet pyTardisDevClient python-dotenv boto3
# src/config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class TardisConfig:
    api_key: str = os.getenv("TARDIS_API_KEY", "")
    exchange: str = os.getenv("TARDIS_EXCHANGE", "binance-futures")
    symbols: list = None
    
    # Parquet 설정
    compression: str = "snappy"  # snappy 또는 zstd
    row_group_size: int = 100_000  # 행 그룹 크기
    engine: str = "pyarrow"  # pyarrow 또는 fastparquet
    
    # 스토리지 설정
    s3_bucket: str = os.getenv("S3_BUCKET", "tardis-data")
    s3_prefix: str = os.getenv("S3_PREFIX", "parquet/binance-futures")
    
    def __post_init__(self):
        if self.symbols is None:
            self.symbols = os.getenv("TARDIS_SYMBOLS", "btcusdt,ethusdt").split(",")

config = TardisConfig()

2단계: Tardis.dev API 연결 및 데이터 수신

# src/connector.py
import asyncio
import json
from typing import AsyncIterator, Dict, Any
from tardis.http import TardisHttpClient
from tardis.interface import Market

class TardisConnector:
    """Tardis.dev API 연결 및 실시간 데이터 스트리밍"""
    
    def __init__(self, config):
        self.config = config
        self.client = None
        self._buffer = []
        self._buffer_size = 10_000  # 버퍼 크기
        
    async def connect(self):
        """API 클라이언트 초기화"""
        self.client = TardisHttpClient(
            exchange=self.config.exchange,
            api_key=self.config.api_key
        )
        await self.client.connect()
        
    async def stream_trades(self, symbol: str) -> AsyncIterator[Dict[str, Any]]:
        """특정 심볼의 트레이드 스트림 수신"""
        if not self.client:
            await self.connect()
            
        market = Market(exchange=self.config.exchange, symbols=[symbol])
        
        async for event in self.client.stream(market):
            if event.type == "trade":
                yield self._normalize_trade_event(event.data)
                
    def _normalize_trade_event(self, event: Dict) -> Dict[str, Any]:
        """트레이드 이벤트 정규화"""
        return {
            "timestamp": event["timestamp"],  # Unix ms
            "symbol": event["symbol"],
            "side": event["side"].upper(),     # BUY/SELL
            "price": float(event["price"]),
            "amount": float(event["amount"]),
            "volume": float(event["price"]) * float(event["amount"]),
            "trade_id": event["id"],
            "is_maker": event.get("is_maker", False)
        }
        
    async def fetch_historical(self, symbol: str, start: int, end: int) -> list:
        """이력 데이터 대량 조회"""
        if not self.client:
            await self.connect()
            
        market = Market(exchange=self.config.exchange, symbols=[symbol])
        trades = []
        
        async for event in self.client.get_historical(
            market, 
            start=start,
            end=end
        ):
            if event.type == "trade":
                trades.append(self._normalize_trade_event(event.data))
                
        return trades


사용 예시

async def example_stream():
    config = TardisConfig()
    connector = TardisConnector(config)
    
    async for trade in connector.stream_trades("btcusdt"):
        print(f"[{trade['timestamp']}] {trade['symbol']}: {trade['price']} x {trade['amount']}")
        # 여기에 Parquet 변환 로직 연결

3단계: JSON → Parquet 변환 및 최적화

# src/transformer.py
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any, Optional

class TardisParquetTransformer:
    """Tardis.dev 데이터를 Parquet로 변환하는 핵심 변환기"""
    
    def

관련 리소스
관련 문서