บทนำ: ทำไมต้องสร้าง ETL Pipeline อัตโนมัติ?

ในโลกของ Data Engineering ยุคปัจจุบัน การจัดการข้อมูลขนาดใหญ่เป็นความท้าทายหลักของทีม Development ทุกคน โดยเฉพาะเมื่อต้องรับมือกับข้อมูลจาก Tardis (แพลตฟอร์มสำหรับ Market Data ชั้นนำ) ที่มีรูปแบบไฟล์หลากหลายและขนาดใหญ่

จากประสบการณ์ตรงของผู้เขียนในการสร้าง Production Pipeline สำหรับองค์กร Fintech ขนาดใหญ่แห่งหนึ่ง พบว่าการประมวลผลข้อมูลแบบ Manual ทำให้เกิดความผิดพลาดซ้ำแล้วซ้ำเล่า และสิ้นเปลืองเวลามากกว่า 40 ชั่วโมงต่อเดือน

สถานการณ์ข้อผิดพลาดจริงที่เคยเจอ

ในการพัฒนาระบบ Tardis ETL Pipeline เวอร์ชันแรก ทีมเผชิญกับปัญหาหลายประการ:

บทความนี้จะอธิบายวิธีแก้ปัญหาทั้งหมดด้วย Architecture ที่ Robust และใช้ HolySheep AI ช่วยในการทำ Data Cleansing อัตโนมัติ

Architecture ภาพรวมของ Tardis ETL Pipeline

┌─────────────────────────────────────────────────────────────────────┐
│                    TARDIS ETL PIPELINE ARCHITECTURE                  │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐      │
│  │ DOWNLOAD │───▶│ EXTRACT  │───▶│ CLEANSE  │───▶│  LOAD    │      │
│  │  (1)     │    │  (2)     │    │  (3)     │    │  (4)     │      │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘      │
│       │              │              │              │                │
│       ▼              ▼              ▼              ▼                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐             │
│  │ S3/Local │  │ GZIP/BZ2 │  │HolySheep │  │PostgreSQL│             │
│  │ Storage  │  │ Streaming│  │AI API    │  │/MongoDB  │             │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘             │
│                                                                     │
│  (1) Retry Logic + Checksum Verification                            │
│  (2) Memory-efficient Streaming Decompression                       │
│  (3) AI-powered Data Validation & Transformation                     │
│  (4) Batch Insert with Transaction Management                       │
└─────────────────────────────────────────────────────────────────────┘

ขั้นตอนที่ 1: การดาวน์โหลดข้อมูล Tardis อัตโนมัติ

การดาวน์โหลดข้อมูลจาก Tardis API ต้องมีระบบ Retry และ Checksum Verification เพื่อป้องกันข้อมูลเสียหาย

# tardis_downloader.py
import requests
import hashlib
import time
from pathlib import Path
from tenacity import retry, stop_after_attempt, wait_exponential

class TardisDownloader:
    """Downloader สำหรับ Tardis Market Data พร้อมระบบ Retry + Checksum"""
    
    BASE_URL = "https://api.tardis.dev/v1"
    CHUNK_SIZE = 8192  # 8KB chunks สำหรับ Streaming
    
    def __init__(self, api_key: str, output_dir: str = "./data/raw"):
        self.api_key = api_key
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def download_with_retry(self, symbol: str, date: str) -> Path:
        """
        ดาวน์โหลดไฟล์ข้อมูลพร้อม Retry Logic
        
        Raises:
            ConnectionError: เมื่อดาวน์โหลดล้มเหลวหลัง Retry 3 ครั้ง
            401 Unauthorized: เมื่อ API Token ไม่ถูกต้อง
        """
        url = f"{self.BASE_URL}/historical/{symbol}/{date}"
        
        try:
            response = self.session.get(
                url, 
                stream=True,
                timeout=(30, 300)  # (connect_timeout, read_timeout)
            )
            response.raise_for_status()
            
            # ตรวจสอบ Status Code
            if response.status_code == 401:
                raise ConnectionError(f"401 Unauthorized: API Token หมดอายุหรือไม่ถูกต้อง")
            elif response.status_code == 429:
                # Rate Limited - รอตาม Retry-After Header
                retry_after = int(response.headers.get("Retry-After", 60))
                time.sleep(retry_after)
                raise ConnectionError("Rate Limited - รอดำเนินการต่อ")
            
            # บันทึกไฟล์พร้อม Streaming
            filename = f"{symbol}_{date}.gz"
            filepath = self.output_dir / filename
            
            with open(filepath, "wb") as f:
                for chunk in response.iter_content(chunk_size=self.CHUNK_SIZE):
                    if chunk:
                        f.write(chunk)
            
            # ตรวจสอบ Checksum (ถ้ามีใน Header)
            expected_checksum = response.headers.get("X-Checksum-SHA256")
            if expected_checksum:
                actual_checksum = self._calculate_sha256(filepath)
                if actual_checksum != expected_checksum:
                    filepath.unlink()  # ลบไฟล์ที่เสียหาย
                    raise ConnectionError(
                        f"Checksum Mismatch: คาดว่า {expected_checksum}, "
                        f"ได้ {actual_checksum}"
                    )
            
            print(f"✅ ดาวน์โหลดสำเร็จ: {filepath}")
            return filepath
            
        except requests.exceptions.Timeout:
            raise ConnectionError(f"Timeout: ไม่สามารถเชื่อมต่อ {url} ภายในเวลาที่กำหนด")
        except requests.exceptions.ConnectionError as e:
            raise ConnectionError(f"ConnectionError: {str(e)}")
    
    def _calculate_sha256(self, filepath: Path) -> str:
        """คำนวณ SHA256 Checksum ของไฟล์"""
        sha256 = hashlib.sha256()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(self.CHUNK_SIZE), b""):
                sha256.update(chunk)
        return sha256.hexdigest()



การใช้งาน

if __name__ == "__main__":
    downloader = TardisDownloader(
        api_key="YOUR_TARDIS_API_KEY",
        output_dir="./data/raw"
    )
    
    try:
        # ดาวน์โหลดข้อมูล BTC/USDT วันที่ 2024-01-15
        filepath = downloader.download_with_retry(
            symbol="binance-futures",
            date="2024-01-15"
        )
        print(f"ไฟล์ถูกบันทึกที่: {filepath}")
        
    except ConnectionError as e:
        print(f"❌ เกิดข้อผิดพลาด: {e}")

ขั้นตอนที่ 2: การแตกไฟล์แบบ Memory-Efficient

การแตกไฟล์ GZIP ขนาดใหญ่ต้องใช้เทคนิค Streaming เพื่อไม่ให้ Memory เต็ม ต่อไปนี้คือโค้ดที่ใช้ใน Production จริง:

# tardis_extractor.py
import gzip
import bz2
import shutil
from pathlib import Path
from typing import Generator, Iterator
import ijson  # Streaming JSON Parser

class MemoryEfficientExtractor:
    """Extractor สำหรับ GZIP/BZ2 พร้อม Streaming แบบไม่กิน Memory"""
    
    CHUNK_SIZE = 65536  # 64KB - Optimal สำหรับ I/O
    
    def __init__(self, output_dir: str = "./data/extracted"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    def extract_gzip_streaming(self, gz_path: Path) -> Path:
        """
        แตกไฟล์ GZIP แบบ Streaming - ไม่โหลดทั้งไฟล์เข้า Memory
        
        ข้อดี:
        - ใช้ Memory คงที่ ไม่ขึ้นกับขนาดไฟล์
        - สามารถประมวลผลไฟล์ขนาด 10GB+ ได้โดยไม่มีปัญหา MemoryError
        """
        output_path = self.output_dir / gz_path.stem  # ลบ .gz ออก
        
        with gzip.open(gz_path, 'rb') as f_in:
            with open(output_path, 'wb') as f_out:
                # Streaming Copy - ไม่ต้องโหลดทั้งไฟล์
                shutil.copyfileobj(f_in, f_out, length=self.CHUNK_SIZE)
        
        print(f"✅ แตกไฟล์สำเร็จ: {output_path}")
        return output_path
    
    def stream_json_lines(self, jsonl_path: Path) -> Generator[dict, None, None]:
        """
        อ่านไฟล์ JSONL แบบ Streaming ด้วย ijson
        
        ตัวอย่างไฟล์ JSONL:
        {"timestamp": "2024-01-15T00:00:00Z", "price": 50000, "volume": 100}
        {"timestamp": "2024-01-15T00:00:01Z", "price": 50001, "volume": 150}
        """
        with open(jsonl_path, 'rb') as f:  # เปิดเป็น Binary mode
            # ใช้ ijson แบบ Streaming
            parser = ijson.items(f, 'item')
            for record in parser:
                yield record
    
    def extract_and_stream(self, gz_path: Path) -> Generator[dict, None, None]:
        """
        Extract + Stream แบบ Combined - ประมวลผลทันทีโดยไม่ต้องบันทึก
        
        เหมาะสำหรับไฟล์ที่ต้องการใช้แค่ครั้งเดียว
        """
        with gzip.open(gz_path, 'rt', encoding='utf-8') as f:
            parser = ijson.items(f, 'item')
            for record in parser:
                yield record



การใช้งานร่วมกับ Pipeline

if __name__ == "__main__":
    extractor = MemoryEfficientExtractor(output_dir="./data/extracted")
    
    # วิธีที่ 1: แตกไฟล์ก่อนแล้วค่อยอ่าน
    extracted = extractor.extract_gzip_streaming(Path("./data/raw/test_data.gz"))
    for record in extractor.stream_json_lines(extracted):
        print(f"Record: {record}")
    
    # วิธีที่ 2: Extract + Stream พร้อมกัน (ประหยัด Disk Space)
    for record in extractor.extract_and_stream(Path("./data/raw/test_data.gz")):
        # ประมวลผลทันที
        process_record(record)

ขั้นตอนที่ 3: Data Cleansing ด้วย HolySheep AI

ขั้นตอนที่สำคัญที่สุดคือการทำความสะอาดข้อมูล (Data Cleansing) ซึ่งใช้ HolySheep AI ในการ:

# tardis_cleanser.py
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
import requests

@dataclass
class CleansingResult:
    """ผลลัพธ์ของการ Cleansing"""
    original_count: int
    cleaned_count: int
    removed_count: int
    modified_count: int
    errors: List[str]

class HolySheepCleanser:
    """Data Cleanser โดยใช้ HolySheep AI API"""
    
    # ✅ ใช้ HolySheep AI - เร็วกว่า 85%, Latency <50ms
    BASE_URL = "https://api.holysheep.ai/v1"  
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def validate_and_clean_batch(
        self, 
        records: List[Dict],
        schema: Optional[Dict] = None
    ) -> CleansingResult:
        """
        Validate และ Clean ข้อมูลทั้ง Batch ด้วย HolySheep AI
        
        Args:
            records: รายการข้อมูลที่ต้องการ Clean
            schema: Schema ที่ต้องการให้ตรวจสอบ (Optional)
        
        Returns:
            CleansingResult: ผลลัพธ์ของการ Cleansing
        """
        result = CleansingResult(
            original_count=len(records),
            cleaned_count=0,
            removed_count=0,
            modified_count=0,
            errors=[]
        )
        
        # แบ่งเป็น Batch ขนาด 100 รายการ (API Limit)
        batch_size = 100
        for i in range(0, len(records), batch_size):
            batch = records[i:i + batch_size]
            
            prompt = self._build_cleansing_prompt(batch, schema)
            
            try:
                cleaned_batch = self._call_holysheep_api(prompt)
                processed = self._process_ai_response(cleaned_batch, batch)
                
                result.cleaned_count += processed["kept"]
                result.removed_count += processed["removed"]
                result.modified_count += processed["modified"]
                
            except Exception as e:
                result.errors.append(f"Batch {i//batch_size}: {str(e)}")
        
        return result
    
    def _build_cleansing_prompt(
        self, 
        records: List[Dict], 
        schema: Optional[Dict]
    ) -> str:
        """สร้าง Prompt สำหรับ AI"""
        
        schema_info = ""
        if schema:
            schema_info = f"\n\nตรวจสอบตาม Schema นี้:\n{json.dumps(schema, indent=2, ensure_ascii=False)}"
        
        prompt = f"""คุณคือ Data Cleansing Specialist
วิเคราะห์และแก้ไขข้อมูลต่อไปนี้:

{json.dumps(records, indent=2, ensure_ascii=False)}{schema_info}

กฎการ Cleansing:
1. ลบ Record ที่มีค่าว่าง (null/None/empty) ในฟิลด์บังคับ
2. แก้ไขรูปแบบวันที่ให้เป็น ISO 8601 (YYYY-MM-DDTHH:MM:SSZ)
3. ตรวจจับ Outliers ในราคาและ Volume (ถ้าราคา < 0 หรือ Volume < 0 ให้ลบ)
4. Standardize สกุลเงิน (USD, THB, EUR)
5. ตรวจสอบ Symbol Format (เช่น BTC/USDT)

ส่งผลลัพธ์เป็น JSON Array พร้อมระบุ:
- kept: รายการที่ถูกต้อง (เรียงตามลำดับเดิม)
- removed: รายการที่ถูกลบ (พร้อมเหตุผล)
- modified: รายการที่ถูกแก้ไข (พร้อมระบุฟิลด์ที่แก้)

Format:
{{"kept": [...], "removed": [...], "modified": [...]}}"""

        return prompt
    
    def _call_holysheep_api(self, prompt: str) -> dict:
        """
        เรียก HolySheep AI API
        
        ✅ ข้อดีของ HolySheep:
        - Latency <50ms (เร็วกว่า OpenAI 85%+)
        - ราคาถูกกว่ามาก
        - รองรับ WeChat/Alipay
        """
        payload = {
            "model": "gpt-4.1",  # $8/MTok - คุ้มค่าที่สุด
            "messages": [
                {"role": "system", "content": "คุณคือ Data Cleansing Specialist ที่เชี่ยวชาญ"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.1,  # ต่ำเพื่อความสม่ำเสมอ
            "response_format": {"type": "json_object"}
        }
        
        response = requests.post(
            f"{self.BASE_URL}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30  # HolySheep เร็วมาก ใช้ Timeout สั้นกว่าได้
        )
        response.raise_for_status()
        
        return response.json()["choices"][0]["message"]["content"]
    
    def _process_ai_response(
        self, 
        ai_response: str, 
        original_batch: List[Dict]
    ) -> dict:
        """ประมวลผล Response จาก AI"""
        try:
            result = json.loads(ai_response)
            return {
                "kept": len(result.get("kept", [])),
                "removed": len(result.get("removed", [])),
                "modified": len(result.get("modified", []))
            }
        except json.JSONDecodeError:
            # ถ้า AI Response ไม่ถูกต้อง คืนค่าเดิมทั้งหมด
            return {
                "kept": len(original_batch),
                "removed": 0,
                "modified": 0
            }



การใช้งาน

if __name__ == "__main__":
    cleanser = HolySheepCleanser(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # ข้อมูลตัวอย่างจาก Tardis
    sample_records = [
        {"symbol": "BTC/USDT", "timestamp": "2024-01-15T00:00:00Z", "price": 50000, "volume": 100},
        {"symbol": "ETH/USDT", "timestamp": "2024-01-15T00:00:01Z", "price": -100, "volume": 50},  # ราคาติดลบ
        {"symbol": "SOL/USDT", "timestamp": "invalid-date", "price": 25.5, "volume": None},  # ข้อมูลไม่ถูกต้อง
    ]
    
    result = cleanser.validate_and_clean_batch(sample_records)
    print(f"ผลลัพธ์: {result}")

ขั้นตอนที่ 4: การนำเข้าฐานข้อมูลแบบ Idempotent

การนำเข้าข้อมูลเข้า Database ต้องมีระบบ Idempotent เพื่อป้องกันข้อมูลซ้ำและรองรับการ Re-run

# tardis_loader.py
import psycopg2
from psycopg2.extras import execute_batch
from datetime import datetime
from typing import List, Dict
from hashlib import sha256
import json

class TardisLoader:
    """Loader สำหรับนำเข้าข้อมูล Tardis เข้า PostgreSQL แบบ Idempotent"""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self._ensure_table_exists()
    
    def _ensure_table_exists(self):
        """สร้าง Table ถ้ายังไม่มี"""
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS tardis_market_data (
            id SERIAL PRIMARY KEY,
            record_hash VARCHAR(64) UNIQUE NOT NULL,  -- Hash สำหรับ Idempotent
            symbol VARCHAR(20) NOT NULL,
            timestamp TIMESTAMPTZ NOT NULL,
            price DECIMAL(20, 8),
            volume DECIMAL(20, 8),
            created_at TIMESTAMPTZ DEFAULT NOW(),
            source_file VARCHAR(255),
            
            -- Index สำหรับ Query เร็ว
            CONSTRAINT unique_record UNIQUE (symbol, timestamp, record_hash)
        );
        
        CREATE INDEX IF NOT EXISTS idx_tardis_symbol_time 
        ON tardis_market_data (symbol, timestamp DESC);
        """
        
        with psycopg2.connect(self.connection_string) as conn:
            with conn.cursor() as cur:
                cur.execute(create_table_sql)
                conn.commit()
    
    def generate_record_hash(self, record: Dict) -> str:
        """สร้าง Hash ที่ Unique สำหรับ Record แต่ละรายการ"""
        # ใช้ Symbol + Timestamp เป็น Key
        hash_input = f"{record['symbol']}|{record['timestamp']}"
        return sha256(hash_input.encode()).hexdigest()
    
    def load_batch_idempotent(
        self, 
        records: List[Dict], 
        source_file: str = None
    ) -> Dict:
        """
        นำเข้าข้อมูลแบบ Idempotent
        
        ข้อดี:
        - รันซ้ำได้หลายครั้งโดยไม่เกิด Duplicate
        - ติดตามได้ว่า Record มาจากไฟล์ไหน
        
        Returns:
            Dict: จำนวน Inserted, Updated, Skipped
        """
        result = {"inserted": 0, "updated": 0, "skipped": 0}
        
        # เตรียมข้อมูลพร้อม Hash
        prepared_records = []
        for record in records:
            prepared = {
                "record_hash": self.generate_record_hash(record),
                "symbol": record["symbol"],
                "timestamp": record["timestamp"],
                "price": record.get("price"),
                "volume": record.get("volume"),
                "source_file": source_file
            }
            prepared_records.append(prepared)
        
        # SQL สำหรับ Insert หรือ Update (Upsert)
        upsert_sql = """
        INSERT INTO tardis_market_data 
            (record_hash, symbol, timestamp, price, volume, source_file)
        VALUES 
            (%(record_hash)s, %(symbol)s, %(timestamp)s, %(price)s, %(volume)s, %(source_file)s)
        ON CONFLICT (record_hash) DO UPDATE SET
            price = EXCLUDED.price,
            volume = EXCLUDED.volume,
            created_at = NOW()
        RETURNING (xmax = 0) AS inserted;
        """
        
        try:
            with psycopg2.connect(self.connection_string) as conn:
                with conn.cursor() as cur:
                    # Execute Batch
                    for record in prepared_records:
                        cur.execute(upsert_sql, record)
                        was_inserted = cur.fetchone()[0]
                        
                        if was_inserted:
                            result["inserted"] += 1
                        else:
                            result["updated"] += 1
                    
                    conn.commit()
                    
        except psycopg2.IntegrityError as e:
            # Duplicate Hash - Skip
            result["skipped"] = len(records)
            print(f"⚠️ Skip {len(records)} records: {e}")
        
        return result
    
    def get_latest_timestamp(self, symbol: str) -> datetime:
        """ดึง Timestamp ล่าสุดของ Symbol เพื่อใช้ในการ Incremental Load"""
        with psycopg2.connect(self.connection_string) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT MAX(timestamp) FROM tardis_market_data WHERE symbol = %s",
                    (symbol,)
                )
                result = cur.fetchone()[0]
                return result or datetime.min



การใช้งานใน Pipeline

if __name__ == "__main__":
    loader = TardisLoader(
        connection_string="postgresql://user:pass

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง