คุณเคยเจอสถานการณ์แบบนี้ไหม? ระบบ data pipeline ที่เคยทำงานราบรื่นมาตลอด 6 เดือน อยู่ๆ ก็เริ่มโยน error มั่วซั่ว: ConnectionError: timeout exceeded 30s ตามมาด้วย IncrementalSyncError: checkpoint corrupted แล้วก็จบด้วย DataLossWarning: 847 records missing ปวดหัวไหมล่ะ? เมื่อวานผมเจอ exactly อย่างนั้นกับ project ที่ใช้ Tardis สำหรับ real-time analytics และวันนี้จะมาแชร์วิธีแก้ที่ค้นพบ
Tardis คืออะไร และทำไมต้องใช้ Incremental Update
Tardis เป็น time-series storage engine ที่ออกแบบมาสำหรับจัดการข้อมูลที่มี timestamp จำนวนมาก โดยมีจุดเด่นเรื่อง 增量更新 (Incremental Update) คือการอัพเดทเฉพาะข้อมูลที่เปลี่ยนแปลงแทนที่จะโหลดทั้งหมดใหม่ ช่วยประหยัด bandwidth และเวลาประมวลผลได้มหาศาล
สถาปัตยกรรม Incremental Sync ใน Tardis
ก่อนจะไปถึงการตั้งค่า มาดูว่า Tardis จัดการ incremental update อย่างไร:
- Checkpoint System — เก็บ cursor position ล่าสุดเพื่อรู้ว่า sync ถึงไหนแล้ว
- Change Detection — ตรวจจับว่า record ไหนมีการเปลี่ยนแปลงผ่าน compare กับ hash หรือ timestamp
- Batch Processing — รวมกลุ่ม updates เพื่อลด I/O operations
- Retry Queue — เก็บ failed operations ไว้ retry ภายหลัง
การตั้งค่า Strategy พื้นฐาน
มาเริ่มต้นด้วยโค้ด Python สำหรับตั้งค่า Tardis incremental sync กันเลย:
import json
from datetime import datetime, timedelta
from tardis_client import TardisClient, TardisReplay
การตั้งค่า Tardis Client
BASE_URL = "https://api.holysheep.ai/v1/tardis" # ตัวอย่าง endpoint
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class TardisIncrementalConfig:
"""คลาสตั้งค่า Incremental Update Strategy สำหรับ Tardis"""
def __init__(self, channel: str, strategy: str = "timestamp"):
self.channel = channel
self.strategy = strategy
self.checkpoint = self._load_checkpoint()
self.retry_count = 3
self.timeout_seconds = 30
self.batch_size = 1000
def _load_checkpoint(self) -> dict:
"""โหลด checkpoint ล่าสุดจากไฟล์"""
try:
with open(f"checkpoint_{self.channel}.json", "r") as f:
return json.load(f)
except FileNotFoundError:
# ถ้าไม่มี checkpoint ให้เริ่มตั้งแต่ 24 ชั่วโมงก่อน
return {
"last_sync": (datetime.now() - timedelta(hours=24)).isoformat(),
"last_id": 0,
"cursor_position": 0
}
def save_checkpoint(self, data: dict):
"""บันทึก checkpoint หลัง sync เสร็จ"""
with open(f"checkpoint_{self.channel}.json", "w") as f:
json.dump(data, f, indent=2)
def get_strategy_config(self) -> dict:
"""กลับค่าการตั้งค่า strategy ตามประเภท"""
strategies = {
"timestamp": {
"field": "updated_at",
"condition": f"> '{self.checkpoint['last_sync']}'",
"order_by": "updated_at ASC"
},
"sequence": {
"field": "id",
"condition": f"> {self.checkpoint['last_id']}",
"order_by": "id ASC"
},
"cursor": {
"field": "cursor_position",
"condition": f"> {self.checkpoint['cursor_position']}",
"order_by": "cursor_position ASC"
}
}
return strategies.get(self.strategy, strategies["timestamp"])
ใช้งาน
config = TardisIncrementalConfig(
channel="stock_prices",
strategy="timestamp"
)
print(f"Loaded checkpoint: {config.checkpoint}")
print(f"Strategy: {config.get_strategy_config()}")Advanced Strategy: Real-time Sync พร้อม Error Handling
ต่อไปเป็นโค้ดที่ใช้งานจริงใน production พร้อมระบบ retry และ graceful shutdown:
import asyncio
import logging
from typing import Optional, Callable, Any
from dataclasses import dataclass
from tardis_client import TardisClient, TardisConnectionException
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SyncResult:
"""ผลลัพธ์ของการ sync"""
records_processed: int
records_failed: int
next_checkpoint: dict
duration_ms: float
class TardisIncrementalSync:
"""ระบบ Incremental Sync พร้อม error handling และ monitoring"""
def __init__(
self,
api_url: str,
api_key: str,
channel: str,
strategy: str = "timestamp",
batch_size: int = 1000,
max_retries: int = 3,
backoff_factor: float = 2.0
):
self.api_url = api_url
self.api_key = api_key
self.channel = channel
self.batch_size = batch_size
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.client = None
self.checkpoint = self._load_checkpoint()
def _load_checkpoint(self) -> dict:
"""โหลด checkpoint จาก storage"""
import json
try:
with open(f"checkpoint_{self.channel}.json", "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {
"last_sync": None,
"last_id": 0,
"cursor_position": 0,
"version": 1
}
async def sync(self, processor: Callable[[list], Any]) -> SyncResult:
"""ทำ incremental sync โดยเรียก processor สำหรับแต่ละ batch"""
import time
start_time = time.time()
records_processed = 0
records_failed = 0
retry_count = 0
backoff = 1.0
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
while retry_count < self.max_retries:
try:
async with aiohttp.ClientSession() as session:
# สร้าง query สำหรับ incremental fetch
query = self._build_incremental_query()
async with session.post(
f"{self.api_url}/replay",
headers=headers,
json=query,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 401:
raise TardisConnectionException(
"Unauthorized: ตรวจสอบ API key ของคุณ"
)
if response.status == 429:
# Rate limited — wait and retry
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning(f"Rate limited, waiting {retry_after}s")
await asyncio.sleep(retry_after)
continue
if response.status != 200:
raise TardisConnectionException(
f"HTTP {response.status}: {await response.text()}"
)
# ประมวลผล response เป็น stream
async for batch in self._iter_batches(response):
try:
await processor(batch)
records_processed += len(batch)
self._update_checkpoint(batch[-1])
except Exception as e:
logger.error(f"Batch processing error: {e}")
records_failed += len(batch)
break # สำเร็จแล้ว ออกจาก loop
except TardisConnectionException as e:
logger.error(f"Connection error: {e}")
retry_count += 1
await asyncio.sleep(backoff)
backoff *= self.backoff_factor
except asyncio.TimeoutError:
logger.error("Connection timeout")
retry_count += 1
await asyncio.sleep(backoff)
backoff *= self.backoff_factor
except Exception as e:
logger.exception(f"Unexpected error: {e}")
raise
return SyncResult(
records_processed=records_processed,
records_failed=records_failed,
next_checkpoint=self.checkpoint,
duration_ms=(time.time() - start_time) * 1000
)
def _build_incremental_query(self) -> dict:
"""สร้าง query สำหรับดึงข้อมูลที่ใหม่กว่า checkpoint"""
from datetime import datetime
last_sync = self.checkpoint.get("last_sync")
return {
"channel": self.channel,
"from": last_sync or datetime.now().isoformat(),
"filter": {
"type": "incremental",
"strategy": self.strategy,
"fields": ["id", "timestamp", "data", "checksum"]
},
"order": "timestamp ASC",
"limit": self.batch_size
}
def _update_checkpoint(self, last_record: dict):
"""อัพเดท checkpoint หลังประมวลผลเสร็จ"""
import json
self.checkpoint = {
"last_sync": last_record.get("timestamp"),
"last_id": last_record.get("id", 0),
"cursor_position": last_record.get("cursor_position", 0),
"version": self.checkpoint.get("version", 1) + 1,
"updated_at": datetime.now().isoformat()
}
with open(f"checkpoint_{self.channel}.json", "w") as f:
json.dump(self.checkpoint, f, indent=2)
ตัวอย่างการใช้งาน
async def main():
sync = TardisIncrementalSync(
api_url="https://api.holysheep.ai/v1/tardis",
api_key="YOUR_HOLYSHEEP_API_KEY",
channel="user_events",
strategy="timestamp",
batch_size=500
)
async def process_batch(batch):
# ประมวลผลแต่ละ batch — ส่งไป storage, analytics, ฯลฯ
print(f"Processing {len(batch)} records")
# TODO: เพิ่ม logic การประมวลผล
result = await sync.sync(process_batch)
print(f"Sync completed: {result.records_processed} processed, {result.records_failed} failed")
print(f"Duration: {result.duration_ms:.2f}ms")
if __name__ == "__main__":
asyncio.run(main())การตั้งค่า Strategy ตามประเภทข้อมูล
ความแตกต่างของ strategy ขึ้นอยู่กับลักษณะข้อมูลของคุณ:
| Strategy | ใช้เมื่อไหร่ | ข้อดี | ข้อเสีย |
|---|---|---|---|
| Timestamp | ข้อมูลมี updated_at field และต้องการ sync แบบ near-realtime | แม่นยำ, รองรับ backfill | ต้องมี index บน timestamp |
| Sequence (ID) | Primary key เป็น auto-increment | เร็วมาก, query ง่าย | ไม่รองรับ update/delete |
| Cursor | ข้อมูลจาก stream ที่มี cursor-based pagination | รองรับ large dataset | ต้องจัดการ cursor อย่างระมัดระวัง |
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ConnectionError: timeout exceeded
อาการ: Sync ดึงข้อมูลมาได้สักพัก แล้วก็เกิด timeout พร้อม error message เต็มๆ ว่า ConnectionError: timeout exceeded 30s
สาเหตุ: Server ตอบสนองช้าเกินไป หรือ query ดึงข้อมูลมากเกินไปในครั้งเดียว
วิธีแก้ไข:
# วิธีที่ 1: เพิ่ม timeout และลด batch_size
sync = TardisIncrementalSync(
api_url="https://api.holysheep.ai/v1/tardis",
api_key="YOUR_HOLYSHEEP_API_KEY",
channel="large_dataset",
batch_size=100, # ลดจาก 1000 เป็น 100
max_retries=5 # เพิ่ม retry attempts
)
วิธีที่ 2: ใช้ streaming response แทน bulk fetch
class StreamingTardisSync(TardisIncrementalSync):
async def _iter_batches(self, response):
"""Iterate response เป็น chunk แทน load ทั้งหมด"""
buffer = []
async for line in response.content:
record = json.loads(line)
buffer.append(record)
if len(buffer) >= self.batch_size:
yield buffer
buffer = []
if buffer:
yield buffer
วิธีที่ 3: เพิ่ม exponential backoff สำหรับ timeout
ดูในโค้ดตัวอย่างด้านบน — backoff_factor = 2.0
2. 401 Unauthorized — Invalid API Key
อาการ: เริ่ม sync ได้ไม่กี่วินาทีก็เจอ 401 Unauthorized: Invalid API key ทั้งๆ ที่ key ถูกต้อง
สาเหตุ: Token หมดอายุ, ใส่ key ผิด format, หรือ API key ไม่มีสิทธิ์ access channel นั้นๆ
วิธีแก้ไข:
# วิธีที่ 1: ตรวจสอบ format และ refresh token
import os
class AuthenticatedTardisSync(TardisIncrementalSync):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._validate_credentials()
def _validate_credentials(self):
"""ตรวจสอบความถูกต้องของ credentials"""
if not self.api_key.startswith("hs_"):
raise ValueError(
"API key ต้องขึ้นต้นด้วย 'hs_' "
"ตรวจสอบที่ https://www.holysheep.ai/dashboard"
)
# ทดสอบ credentials ด้วย lightweight request
import requests
test_resp = requests.get(
f"{self.api_url}/health",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=5
)
if test_resp.status_code == 401:
raise ValueError(
"API key ไม่ถูกต้องหรือหมดอายุ "
"กรุณาสร้างใหม่ที่ https://www.holysheep.ai/dashboard"
)
def _get_auth_headers(self) -> dict:
"""สร้าง headers พร้อม authentication"""
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Channel": self.channel,
"X-Client-Version": "tardis-sync-v2.0"
}
วิธีที่ 2: ใช้ environment variable แทน hardcode
API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if not API_KEY:
raise EnvironmentError(
"กรุณาตั้งค่า HOLYSHEEP_API_KEY ใน environment variables"
)3. IncrementalSyncError: checkpoint corrupted
อาการ: วันรุ่งขึ้นมาทำ sync ต่อ แต่ได้ error IncrementalSyncError: checkpoint corrupted at position 847293
สาเหตุ: ไฟล์ checkpoint ถูกเขียนทับขณะ sync กำลังทำงาน หรือ format ไม่ตรงกัน
วิธีแก้ไข:
import json
import fcntl
from pathlib import Path
from datetime import datetime
class SafeCheckpointManager:
"""จัดการ checkpoint อย่างปลอดภัยด้วย file locking"""
def __init__(self, channel: str, checkpoint_dir: str = "./checkpoints"):
self.channel = channel
self.checkpoint_path = Path(checkpoint_dir) / f"{channel}.json"
self.checkpoint_path.parent.mkdir(parents=True, exist_ok=True)
def load(self) -> dict:
"""โหลด checkpoint พร้อมตรวจสอบความถูกต้อง"""
if not self.checkpoint_path.exists():
return self._default_checkpoint()
with open(self.checkpoint_path, "r+") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_SH) # Shared lock for reading
try:
data = json.load(f)
# ตรวจสอบ schema version
if "version" not in data:
# Migrate จาก schema เก่า
data = self._migrate_checkpoint(data)
# ตรวจสอบความสมบูรณ์
required_fields = ["last_sync", "last_id", "cursor_position"]
for field in required_fields:
if field not in data:
raise ValueError(f"Checkpoint missing required field: {field}")
return data
except (json.JSONDecodeError, ValueError) as e:
# Recovery: ใช้ checkpoint ก่อนหน้าหรือสร้างใหม่
return self._recover_from_corruption(e)
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
def save(self, data: dict):
"""บันทึก checkpoint อย่างปลอดภัยด้วย atomic write"""
temp_path = self.checkpoint_path.with_suffix(".tmp")
with open(temp_path, "w") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX) # Exclusive lock for writing
try:
# เพิ่ม metadata
data["version"] = data.get("version", 0) + 1
data["saved_at"] = datetime.now().isoformat()
data["checksum"] = self._calculate_checksum(data)
json.dump(data, f, indent=2)
f.flush()
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
# Atomic rename
temp_path.replace(self.checkpoint_path)
def _default_checkpoint(self) -> dict:
"""สร้าง checkpoint เริ่มต้น"""
return {
"last_sync": None,
"last_id": 0,
"cursor_position": 0,
"version": 1,
"created_at": datetime.now().isoformat()
}
def _migrate_checkpoint(self, old_data: dict) -> dict:
"""ย้ายข้อมูลจาก schema เก่า"""
return {
"last_sync": old_data.get("timestamp"),
"last_id": old_data.get("last_id", 0),
"cursor_position": old_data.get("position", 0),
"version": 1,
"migrated_from": "v0"
}
def _recover_from_corruption(self, error: Exception) -> dict:
"""กู้คืนจาก checkpoint ที่เสียหาย"""
import shutil
# Rename ไฟล์เสียเป็น backup
backup_path = self.checkpoint_path.with_suffix(".corrupted")
self.checkpoint_path.rename(backup_path)
print(f"Checkpoint corrupted: {error}")
print(f"Backup saved to: {backup_path}")
return self._default_checkpoint()
def _calculate_checksum(self, data: dict) -> str:
"""คำนวณ checksum สำหรับตรวจสอบความถูกต้อง"""
import hashlib
content = json.dumps(data, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]4. DataLossWarning: records missing after sync
อาการ: Sync เสร็จสมบูรณ์ แต่ analytics dashboard แสดงว่ามี records หายไป พร้อม warning DataLossWarning: 847 records missing between ID 1000-5000
สาเหตุ: ข้อมูลถูกลบหรือ update ระหว่าง sync ก่อนที่ checkpoint จะถูกบันทึก
วิธีแก้ไข:
import hashlib
from typing import Set, Tuple
class IntegrityCheckingSync(TardisIncrementalSync):
"""Sync พร้อมตรวจสอบความสมบูรณ์ของข้อมูล"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.expected_count = 0
self.seen_ids: Set[int] = set()
self.duplicates: Set[int] = set()
async def sync(self, processor: Callable) -> Tuple[SyncResult, dict]:
"""Sync พร้อม integrity check"""
result = await super().sync(processor)
integrity_report = self._generate_integrity_report()
if integrity_report["gaps"] or integrity_report["duplicates"]:
self._log_integrity_warning(integrity_report)
return result, integrity_report
def _generate_integrity_report(self) -> dict:
"""สร้างรายงานความสมบูรณ์ของข้อมูล"""
gaps = self._find_id_gaps()
return {
"total_seen": len(self.seen_ids),
"duplicates": len(self.duplicates),
"gaps": gaps,
"has_data_loss": bool(gaps),
"integrity_score": self._calculate_integrity_score()
}
def _find_id_gaps(self) -> list:
"""หา ID ที่หายไประหว่าง sync"""
if not self.seen_ids:
return []
sorted_ids = sorted(self.seen_ids)
gaps = []
for i in range(len(sorted_ids) - 1):
current = sorted_ids[i]
next_id = sorted_ids[i + 1]
if next_id - current > 1:
gaps.append({
"start": current + 1,
"end": next_id - 1,
"missing_count": next_id - current - 1
})
return gaps
def _calculate_integrity_score(self) -> float:
"""คำนวณคะแนนความสมบูรณ์ (0-100)"""
if not self.expected_count:
return 100.0
unique_count = len(self.seen_ids)
duplicate_count = len(self.duplicates)
missing_count = self.expected_count - unique_count
score = (unique_count / self.expected_count) * 100
score -= (duplicate_count * 0.5) # หักคะแนน duplicate
return max(0.0, min(100.0, score))
def _log_integrity_warning(self, report: dict):
"""บันทึก warning เมื่อพบความผิดปกติ"""
message = [
"⚠️ Data Integrity Warning",
f"Integrity Score: {report['integrity_score']:.1f}%",
f"Seen: {report['total_seen']} records"
]
if report["gaps"]:
for gap in report["gaps"][:3]: # แสดงแค่ 3 gaps แรก
message.append(
f" Gap: IDs {gap['start']}-{gap['end']} "
f"({gap['missing_count']} records missing)"
)
if report["duplicates"]:
message.append(f"⚠️ Duplicates: {report['duplicates']} records")
logger.warning("\n".join(message))
# ส่ง alert ไปยัง monitoring system
self._send_alert(report)