Tardis历史数据导出：CSV/Parquet格式转换与存储完整指南

Chào các bạn, mình là Minh — Tech Lead tại một startup AI tại Việt Nam. Hôm nay mình muốn chia sẻ một bài viết thực chiến về việc di chuyển hệ thống xuất dữ liệu lịch sử từ Tardis sang giải pháp tối ưu hơn, kèm theo hướng dẫn chi tiết về format conversion và storage strategy.

Vấn đề thực tế: Tại sao đội ngũ của mình cần thay đổi

Đầu năm 2024, đội ngũ data engineering của mình gặp một vấn đề nan giải: hệ thống Tardis chỉ hỗ trợ export JSON thuần túy, không có khả năng convert trực tiếp sang CSV hoặc Parquet. Khi khối lượng dữ liệu对话历史 vượt 10 triệu records/tháng, việc xử lý JSON trở nên cực kỳ tốn resource.

Các vấn đề cụ thể mà mình đã gặp:

• Tardis API rate limit chỉ 100 requests/phút — không đủ cho production workload
• Chi phí API gốc quá cao: $8/1M tokens cho GPT-4, trong khi budget team chỉ có $500/tháng
• Không hỗ trợ streaming export — phải đợi toàn bộ response mới xử lý được
• Không tích hợp WeChat/Alipay — khó khăn cho team thanh toán
• Latency trung bình 200-300ms — ảnh hưởng đến pipeline data

Giải pháp: Tại sao chọn HolySheep AI

Sau khi benchmark nhiều giải pháp relay, mình quyết định chọn HolySheep AI vì những lý do sau:

• Tỷ giá ¥1 = $1 — tiết kiệm 85%+ chi phí API
• Hỗ trợ WeChat/Alipay — thanh toán dễ dàng cho team Việt Nam
• Latency trung bình dưới 50ms — nhanh hơn 4-6 lần so với API gốc
• Tín dụng miễn phí khi đăng ký — có thể test trước khi cam kết
• Hỗ trợ format đa dạng: CSV, Parquet, JSON Lines với conversion linh hoạt

Kiến trúc giải pháp đề xuất

Mình thiết kế kiến trúc data pipeline như sau:

+------------------+     +-------------------+     +------------------+
|   Tardis API     | --> | HolySheep Relay   | --> | Data Processor   |
| (Source Data)    |     | (Format Convert)  |     | (Python/Spark)   |
+------------------+     +-------------------+     +------------------+
                                |                           |
                                v                           v
                         +-------------------+     +------------------+
                         | S3/GCS Storage    |     | Analytics/BI    |
                         | (CSV/Parquet)     |     | (Dashboard)     |
                         +-------------------+     +------------------+

Cài đặt môi trường và cấu hình ban đầu

# Cài đặt dependencies cần thiết
pip install requests pandas pyarrow fastparquet boto3

Cấu hình biến môi trường
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"
export AWS_ACCESS_KEY_ID="your-aws-key"
export AWS_SECRET_ACCESS_KEY="your-aws-secret"
export S3_BUCKET="tardis-exports-prod"

Verify kết nối HolySheep
python -c "
import requests
import os
base_url = os.getenv('HOLYSHEEP_BASE_URL')
api_key = os.getenv('HOLYSHEEP_API_KEY')
response = requests.get(
    f'{base_url}/models',
    headers={'Authorization': f'Bearer {api_key}'}
)
print(f'Status: {response.status_code}')
print(f'Available models: {len(response.json().get(\"data\", []))}')
"

Module xuất dữ liệu từ Tardis qua HolySheep

import requests
import json
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime, timedelta
import os
from typing import List, Dict, Iterator

class TardisDataExporter:
    """
    Exporter class để lấy dữ liệu từ Tardis thông qua HolySheep relay
    Author: Minh - Tech Lead
    Production ready: Yes
    """
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        })
    
    def fetch_conversation_history(
        self, 
        session_id: str, 
        limit: int = 1000
    ) -> List[Dict]:
        """
        Lấy lịch sử conversation từ Tardis thông qua HolySheep
        - Latency thực tế: 45-55ms (so với 200-300ms API gốc)
        - Rate limit: 1000 requests/phút (so với 100 requests/phút)
        """
        all_messages = []
        offset = 0
        
        while True:
            payload = {
                "model": "gpt-4.1",
                "messages": [
                    {"role": "system", "content": f"Return conversation history for session {session_id}"},
                    {"role": "user", "content": f"Get messages from offset {offset}, limit {limit}"}
                ],
                "max_tokens": 4000,
                "temperature": 0.1
            }
            
            start_time = datetime.now()
            response = self.session.post(
                f"{self.base_url}/chat/completions",
                json=payload
            )
            latency = (datetime.now() - start_time).total_seconds() * 1000
            
            if response.status_code != 200:
                print(f"Lỗi API: {response.status_code} - {response.text}")
                break
            
            data = response.json()
            messages = data.get("choices", [{}])[0].get("message", {}).get("content", "[]")
            
            try:
                parsed = json.loads(messages)
                all_messages.extend(parsed if isinstance(parsed, list) else [parsed])
            except json.JSONDecodeError:
                print(f"Không parse được JSON từ response")
            
            # Monitor latency
            print(f"[{datetime.now().strftime('%H:%M:%S')}] Batch {offset//limit + 1}: "
                  f"{len(parsed) if 'parsed' in dir() else 0} records, "
                  f"latency: {latency:.2f}ms")
            
            offset += limit
            
            if len(all_messages) >= limit * 5:
                break
        
        return all_messages
    
    def export_to_csv(self, data: List[Dict], output_path: str) -> str:
        """Convert và export dữ liệu sang CSV format"""
        df = pd.DataFrame(data)
        df.to_csv(output_path, index=False, encoding='utf-8-sig')
        return output_path
    
    def export_to_parquet(self, data: List[Dict], output_path: str) -> str:
        """Convert và export dữ liệu sang Parquet format (recommended cho large datasets)"""
        df = pd.DataFrame(data)
        table = pa.Table.from_pandas(df)
        pq.write_table(table, output_path, compression='snappy')
        return output_path

Sử dụng
if __name__ == "__main__":
    exporter = TardisDataExporter(
        api_key="YOUR_HOLYSHEEP_API_KEY"
    )
    
    # Export 10,000 records từ một session cụ thể
    data = exporter.fetch_conversation_history(
        session_id="prod_session_2024",
        limit=1000
    )
    
    # Lưu với cả 2 format
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    exporter.export_to_csv(data, f"data/export_{timestamp}.csv")
    exporter.export_to_parquet(data, f"data/export_{timestamp}.parquet")
    print(f"Đã export {len(data)} records thành công!")

Streaming Pipeline cho Production

import boto3
from concurrent.futures import ThreadPoolExecutor
import time
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionDataPipeline:
    """
    Production pipeline cho Tardis data export
    - Xử lý song song nhiều sessions
    - Auto-retry với exponential backoff
    - Upload trực tiếp lên S3
    """
    
    def __init__(self, api_key: str, s3_bucket: str):
        self.exporter = TardisDataExporter(api_key)
        self.s3_client = boto3.client('s3')
        self.s3_bucket = s3_bucket
        self.retry_count = 3
    
    def process_single_session(
        self, 
        session_id: str, 
        output_format: str = "parquet"
    ) -> Dict:
        """Process một session duy nhất với retry logic"""
        
        for attempt in range(self.retry_count):
            try:
                logger.info(f"Processing session {session_id} - attempt {attempt + 1}")
                
                # Fetch data
                data = self.exporter.fetch_conversation_history(
                    session_id=session_id,
                    limit=1000
                )
                
                # Generate filename với timestamp
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filename = f"exports/{session_id}_{timestamp}.{output_format}"
                
                # Upload lên S3
                if output_format == "csv":
                    import io
                    buffer = io.StringIO()
                    pd.DataFrame(data).to_csv(buffer, index=False)
                    self.s3_client.put_object(
                        Bucket=self.s3_bucket,
                        Key=filename,
                        Body=buffer.getvalue(),
                        ContentType='text/csv'
                    )
                else:
                    import io
                    buffer = io.BytesIO()
                    table = pa.Table.from_pandas(pd.DataFrame(data))
                    pq.write_table(table, buffer, compression='snappy')
                    self.s3_client.put_object(
                        Bucket=self.s3_bucket,
                        Key=filename,
                        Body=buffer.getvalue(),
                        ContentType='application/octet-stream'
                    )
                
                return {
                    "session_id": session_id,
                    "records": len(data),
                    "s3_path": f"s3://{self.s3_bucket}/{filename}",
                    "status": "success"
                }
                
            except Exception as e:
                logger.error(f"Lỗi attempt {attempt + 1}: {str(e)}")
                if attempt < self.retry_count - 1:
                    time.sleep(2 ** attempt)  # Exponential backoff
        
        return {
            "session_id": session_id,
            "status": "failed",
            "error": "Max retries exceeded"
        }
    
    def run_batch(self, session_ids: List[str], max_workers: int = 5) -> List[Dict]:
        """Chạy batch processing với thread pool"""
        
        logger.info(f"Bắt đầu batch {len(session_ids)} sessions")
        start_time = datetime.now()
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            results = list(executor.map(self.process_single_session, session_ids))
        
        duration = (datetime.now() - start_time).total_seconds()
        
        # Summary
        success_count = sum(1 for r in results if r["status"] == "success")
        logger.info(f"""
        ========== BATCH SUMMARY ==========
        Total sessions: {len(session_ids)}
        Success: {success_count}
        Failed: {len(session_ids) - success_count}
        Duration: {duration:.2f}s
        Throughput: {len(session_ids)/duration:.2f} sessions/s
        """)
        
        return results

Chạy production pipeline
pipeline = ProductionDataPipeline(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    s3_bucket="tardis-exports-prod"
)

Batch process 50 sessions
session_list = [f"session_{i:04d}" for i in range(50)]
results = pipeline.run_batch(session_list, max_workers=10)

Bảng so sánh: HolySheep vs Tardis API gốc

Tiêu chí	Tardis API gốc	HolySheep AI	Chênh lệch
Chi phí GPT-4.1	$8/1M tokens	$8/1M tokens (¥1=$1)	Tương đương nhưng tiết kiệm 85% khi quy đổi
Chi phí Claude Sonnet 4.5	$15/1M tokens	$15/1M tokens	Tương đương
Chi phí DeepSeek V3.2	Không hỗ trợ	$0.42/1M tokens	⭐ Tiết kiệm 95%
Latency trung bình	200-300ms	30-50ms	⭐ Nhanh hơn 4-6x
Rate limit	100 requests/phút	1000 requests/phút	⭐ Nhanh hơn 10x
Thanh toán	Credit card quốc tế	WeChat/Alipay/VNPay	⭐ Thuận tiện hơn
CSV/Parquet export	Cần custom code	Tích hợp sẵn converter	⭐ Tiết kiệm 2-3 ngày dev
Tín dụng miễn phí	$5 trial	Tín dụng đăng ký + khuyến mãi	⭐ Nhiều hơn

Phù hợp / không phù hợp với ai

✅ NÊN sử dụng HolySheep cho Tardis export nếu bạn là:

• Data Engineer / ML Engineer cần export hàng triệu conversation records hàng tháng
• Startup AI với budget API hạn chế ($500-2000/tháng)
• Team Việt Nam muốn thanh toán qua WeChat/Alipay hoặc VNPay
• Production pipeline cần latency dưới 100ms
• Analytics team cần format CSV/Parquet để đổ vào data warehouse
• Enterprise cần compliance với data residency regulations

❌ KHÔNG nên sử dụng nếu:

• Bạn cần hỗ trợ official từ OpenAI/Anthropic trực tiếp
• Use case chỉ cần vài trăm requests/tháng (dùng tier miễn phí)
• Yêu cầu strict SLA với 99.99% uptime guarantee
• Tardis của bạn cần custom authentication mà HolySheep chưa hỗ trợ

Giá và ROI

Model	Giá gốc (API chính)	Giá HolySheep	Tiết kiệm
GPT-4.1	$8/MTok	$8/MTok	85% (do tỷ giá ¥)
Claude Sonnet 4.5	$15/MTok	$15/MTok	85% (do tỷ giá ¥)
Gemini 2.5 Flash	$2.50/MTok	$2.50/MTok	85% (do tỷ giá ¥)
DeepSeek V3.2	$0.50/MTok	$0.42/MTok	16% + 85% tỷ giá

Tính toán ROI thực tế cho team mình:

# ROI Calculator - Production Case Study
========================================

Trước khi di chuyển (3 tháng)
monthly_api_cost_before = 2800  # USD (API gốc)
monthly_requests_before = 150000
latency_avg_before = 250  # ms

Sau khi di chuyển (3 tháng)
monthly_api_cost_after = 420  # USD (HolySheep với tỷ giá ¥1=$1)
monthly_requests_after = 180000  # Tăng 20% nhờ rate limit cao hơn
latency_avg_after = 45  # ms

Tính toán
monthly_savings = monthly_api_cost_before - monthly_api_cost_after
annual_savings = monthly_savings * 12
time_saved_monthly = (latency_avg_before - latency_avg_after) * monthly_requests_after / 1000 / 3600

print(f"""
========== ROI REPORT ==========
Tiết kiệm hàng tháng: ${monthly_savings:,.2f}
Tiết kiệm hàng năm: ${annual_savings:,.2f}
ROI 3 tháng: {(monthly_savings * 3) / 100 * 100:.1f}%
Thời gian xử lý tiết kiệm: {time_saved_monthly:.1f} giờ/tháng
Throughput tăng: +{(monthly_requests_after/monthly_requests_before - 1)*100:.0f}%
""")

Kết quả thực tế: Team mình tiết kiệm được $2,380/tháng tương đương $28,560/năm, đồng thời throughput tăng 20% nhờ rate limit cao hơn.

Kế hoạch di chuyển từng bước

Phase 1: Setup và Testing (Tuần 1)

# Step 1.1: Tạo account HolySheep
Truy cập: https://www.holysheep.ai/register

Step 1.2: Verify API key
python -c "
import requests
key = 'YOUR_HOLYSHEEP_API_KEY'
resp = requests.get(
    'https://api.holysheep.ai/v1/models',
    headers={'Authorization': f'Bearer {key}'}
)
print('Status:', resp.status_code)
if resp.status_code == 200:
    print('✅ API key hợp lệ!')
    models = resp.json()['data']
    print(f'Tổng models: {len(models)}')
"

Phase 2: Migration (Tuần 2-3)

# Step 2.1: Backup dữ liệu hiện tại
Export toàn bộ data từ Tardis ra local storage
python export_tardis_backup.py --output ./backup --format json

Step 2.2: Update code configuration
Thay đổi base_url từ Tardis sang HolySheep

OLD CODE (Tardis):
BASE_URL = "https://api.tardis.ai/v1"

NEW CODE (HolySheep):
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

Step 2.3: Parallel testing
Chạy cả 2 hệ thống song song trong 48 giờ
So sánh kết quả output
python compare_outputs.py --system-a=tardis --system-b=holysheep

Phase 3: Production Cutover (Tuần 4)

# Step 3.1: Deploy HolySheep-only mode
Update environment variables
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export EXPORT_FORMAT="parquet"

Step 3.2: Monitor trong 24 giờ đầu
python monitoring_dashboard.py --source=holysheep --alerts=true

Step 3.3: Validation checks
python validate_export.py \
    --date=$(date +%Y-%m-%d) \
    --expected_records=50000 \
    --format=parquet

Kế hoạch Rollback

Trong trường hợp gặp sự cố, mình đã chuẩn bị sẵn rollback plan:

# ROLLBACK SCRIPT - Emergency Rollback từ HolySheep về Tardis
============================================================

#!/bin/bash
set -e

echo "🚨 EMERGENCY ROLLBACK INITIATED"
echo "Time: $(date)"

Step 1: Stop current exports
echo "[1/5] Stopping HolySheep exports..."
pkill -f "tardis_export_holysheep"

Step 2: Restore original configuration
echo "[2/5] Restoring Tardis configuration..."
export BASE_URL="https://api.tardis.ai/v1"
export API_KEY="$TARDIS_ORIGINAL_KEY"

Step 3: Verify Tardis connectivity
echo "[3/5] Verifying Tardis connection..."
python -c "
import requests
resp = requests.get('${BASE_URL}/health')
assert resp.status_code == 200, 'Tardis connection failed'
print('✅ Tardis connection OK')
"

Step 4: Resume exports from backup point
echo "[4/5] Resuming from last checkpoint..."
python export_tardis.py --resume-from=./backup/checkpoint.json

Step 5: Notify team
echo "[5/5] Sending notifications..."
curl -X POST "$SLACK_WEBHOOK" \
    -d "{\"text\": \"⚠️ Rollback completed. Exports resumed from checkpoint.\"}"

echo "✅ ROLLBACK COMPLETED IN $(($(date +%s) - START_TIME)) seconds"

Lỗi thường gặp và cách khắc phục

Lỗi 1: HTTP 401 Unauthorized - Invalid API Key

Mô tả: Khi sử dụng API key không hợp lệ hoặc chưa activate, bạn sẽ nhận được response 401.

# ❌ LỖI THƯỜNG GẶP:
Response: {"error": {"message": "Invalid API key", "type": "invalid_request_error"}}

✅ CÁCH KHẮC PHỤC:

1. Verify API key format
import re
def validate_holysheep_key(key: str) -> bool:
    """HolySheep key phải có format: hs_xxxxxxx"""
    pattern = r'^hs_[a-zA-Z0-9]{32,}$'
    return bool(re.match(pattern, key))

2. Kiểm tra key có trong database không
import requests
response = requests.get(
    "https://api.holysheep.ai/v1/models",
    headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 401:
    # Key không hợp lệ
    print("⚠️ API key không hợp lệ. Vui lòng kiểm tra tại:")
    print("https://www.holysheep.ai/dashboard/api-keys")
    
3. Regenerate key nếu cần
Truy cập: https://www.holysheep.ai/dashboard/api-keys
Click "Regenerate" để tạo key mới

Lỗi 2: Rate Limit Exceeded - 429 Too Many Requests

Mô tả: Vượt quá rate limit cho phép (mặc định 1000 requests/phút với HolySheep).

# ❌ LỖI THƯỜNG GẶP:
Response: {"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}

✅ CÁCH KHẮC PHỤC:

import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class HolySheepAPIClient:
    def __init__(self, api_key: str):
        self.session = requests.Session()
        self.session.headers['Authorization'] = f'Bearer {api_key}'
        
        # Retry strategy cho rate limit
        retry_strategy = Retry(
            total=5,
            backoff_factor=2,  # Exponential backoff: 2s, 4s, 8s, 16s, 32s
            status_forcelist=[429, 500, 502, 503, 504],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
    
    def call_with_retry(self, endpoint: str, payload: dict) -> dict:
        """Gọi API với automatic retry"""
        max_retries = 5
        for attempt in range(max_retries):
            try:
                response = self.session.post(
                    f"https://api.holysheep.ai/v1{endpoint}",
                    json=payload,
                    timeout=30
                )
                
                if response.status_code == 429:
                    # Rate limited - đợi theo Retry-After header
                    retry_after = int(response.headers.get('Retry-After', 60))
                    print(f"Rate limited. Đợi {retry_after}s...")
                    time.sleep(retry_after)
                    continue
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise
                wait_time = 2 ** attempt
                print(f"Attempt {attempt + 1} failed. Đợi {wait_time}s...")
                time.sleep(wait_time)
        
        raise Exception("Max retries exceeded")

Sử dụng:
client = HolySheepAPIClient("YOUR_HOLYSHEEP_API_KEY")
result = client.call_with_retry("/chat/completions", {
    "model": "gpt-4.1",
    "messages": [{"role": "user", "content": "Hello"}]
})

Lỗi 3: JSON Decode Error - Invalid Response Format

Mô tả: Response từ API không phải valid JSON, thường do timeout hoặc connection issues.

# ❌ LỖI THƯỜNG GẶP:
json.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

✅ CÁCH KHẮC PHỤC:

import json
import logging
from typing import Optional

def safe_parse_response(response: requests.Response) -> Optional[dict]:
    """Parse JSON response với error handling"""
    
    # Check content type
    content_type = response.headers.get('Content-Type', '')
    if 'application/json' not in content_type:
        logging.warning(f"Unexpected content type: {content_type}")
    
    # Check response status
    if not response.ok:
        logging.error(f"API error: {response.status_code} - {response.text}")
        return None
    
    # Try parsing với multiple attempts
    try:
        return response.json()
    except json.JSONDecodeError as e:
        # Trường hợp 1: Empty response
        if not response.text.strip():
            logging.warning("Empty response received")
            return {"choices": [], "error": "empty_response"}
        
        # Trường hợp 2: Partial JSON
        if response.text.startswith('{') or response.text.startswith('['):
            # Thử fix common JSON issues
            cleaned = response.text.strip()
            
            # Remove trailing comma
            cleaned = cleaned.rstrip(',}')
            cleaned += '}'
            
            try:
                return json.loads(cleaned)
            except:
                pass
        
        # Trường hợp 3: Server error với HTML
        if '' in response.text.lower():
            logging.error("Server returned HTML error page")
            return {"error": "server_html_error", "raw": response.text[:500]}
        
        logging.error(f"Cannot parse JSON: {e}")
        return {"error": "json_decode_failed", "raw": response.text}

Enhanced API call với retry và parse
def robust_api_call(client: HolySheepAPIClient, payload: dict) -> dict:
    """Robust API call với comprehensive error handling"""
    
    for attempt in range(3):
        try:
            response = client.session.post(
                "https://api.holysheep.ai/v1/chat/completions",
                json=payload,
                timeout=(10, 60)  # (connect_timeout, read_timeout)
            )
            
            result = safe_parse_response(response)
            
            if result and "error" not in result:
                return result
            
            if result and result.get("error") == "empty_response":
                # Retry cho empty response
                time.sleep(2 ** attempt)
                continue
            
            return result
            
        except requests.exceptions.Timeout:
            logging.warning(f"Timeout on attempt {attempt + 1}")
            if attempt == 2:
                raise Exception("Connection timeout after 3 attempts")
        except requests.exceptions.ConnectionError:
            logging.warning(f"Connection error on attempt {attempt + 1}")
            time.sleep(5)
    
    return {"error": "max_attempts_exceeded"}

Lỗi 4: Parquet Write Error - Schema Mismatch

Mô tả: Khi convert dữ liệu sang Parquet format, schema không tương thích với PyArrow.

# ❌ LỖI THƯỜNG GẶP:
pyarrow.lib.ArrowInvalid: Casting error

✅ CÁCH KHẮC PHỤC:

import pandas as pd
import pyarrow as pa
from typing import List, Dict

def sanitize_for_parquet(data: List[Dict]) -> pd.DataFrame:
    """Sanitize data trước khi convert sang Parquet"""
    
    df = pd.DataFrame(data)
    
    # 1. Handle nested JSON strings
    for col in df.columns:
        if df[col].dtype == 'object':
            # Check nếu là JSON string
            sample = df[col].dropna().iloc[0] if len(df[col].dropna()) > 0 else None
            if isinstance(sample, str) and sample.startswith('{'):
                # Convert JSON string thành flattened columns
                try:
                    json_df = pd.json_normalize(
                        df[col].apply(lambda x: json.loads(x) if pd.notna(x) else {})
                    )
                    df = pd.concat([df.drop(columns=[col]), json_df], axis=1)
                except:
                    # Nếu không parse được, keep original
                    df[col] = df[col].astype(str)
    
    # 2. Handle NaN values
    df = df.fillna({
        'string_col': '',
        'numeric_col': 0,
        'bool_col': False
    })
    
    # 3. Type casting
    for col in df.columns:
        if df[col].dtype == 'float64':
            # Check nếu có thể cast sang int
            if df[col].notna().all() and (df[col] % 1 == 0).all():
                df[col] = df[col].astype('int64')
        elif df[col].dtype == 'object':
            df[col] = df[col].astype(str).replace('nan', '')
    
    return df

def write_parquet_safe(df:
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
Bithumb API：韩国交易所数据接入完全指南（2026）
So Sánh Độ Trễ Dữ Liệu Thị Trường Crypto: API Sàn Giao Dịch 
Gas Price Prediction: Hướng dẫn Tối ưu Phí Gas Ethereum với