Tardis L2 สแนปชอต: การลดต้นทุนจัดเก็บข้อมูล 85% จาก WebSocket สู่ Parquet Data Lake

บทนำ: ทำไมการจัดเก็บ L2 สแนปชอตถึงกลายเป็นภาระค่าใช้จ่าย

ในโลกของ AI และ Data Engineering ปี 2026 การจัดการ L2 (Layer 2) สแนปชอต กลายเป็นความท้าทายสำคัญ โดยเฉพาะสำหรับระบบที่ต้องการความยืดหยุ่นในการ replay ข้อมูล เช่น ระบบ RAG (Retrieval-Augmented Generation) ขององค์กร หรือระบบ AI ลูกค้าสัมพันธ์ที่ต้องวิเคราะห์ historical conversation ผมเคยทำงานกับทีมหนึ่งที่รันระบบ Tardis สำหรับจัดเก็บ conversation logs ของ AI chatbot ระดับ enterprise ในช่วงแรก เราใช้ WebSocket streaming แบบดั้งเดิม ซึ่งส่งผลให้ค่าใช้จ่ายด้าน storage พุ่งสูงถึง **$4,200/เดือน** สำหรับข้อมูลเพียง 45GB หลังจากปรับ architecture มาใช้ Parquet data lake เราลดต้นทุนลงเหลือ **$630/เดือน** สำหรับข้อมูล 280GB ที่ queryable ได้รวดเร็ว — **ประหยัดได้ 85%** พร้อมประสิทธิภาพที่ดีกว่าเดิมหลายเท่า บทความนี้จะพาคุณไปดูว่า ทำไม WebSocket ไม่เหมาะกับการจัดเก็บระยะยาว ทำไม Parquet ถึงเป็นคำตอบ และขั้นตอนการ migrate อย่างละเอียด

L2 สแนปชอตคืออะไร: พื้นฐานที่ควรเข้าใจ

L2 สแนปชอต ในบริบทของ Tardis หมายถึง สำเนาของข้อมูล state ณ ช่วงเวลาหนึ่ง ที่เก็บรวบรวมระดับ "message cluster" ไม่ใช่แค่ individual messages ทำให้ระบบสามารถ: - **Replay การสนทนา** ไปยังจุดใดก็ได้ในอดีต - **Debug ปัญหา** โดยดู full context ของ session - **Fine-tune โมเดล** จาก conversation patterns ที่ผ่านมา - **วิเคราะห์ behavior** ของผู้ใช้แบบ historical ปัญหาคือ ข้อมูลประเภทนี้มีขนาดใหญ่มาก โดยเฉพาะเมื่อรวม metadata, embeddings และ context windows ทั้งหมด

ปัญหา WebSocket แบบดั้งเดิม

โครงสร้างต้นทุนที่ไม่ยั่งยืน

WebSocket streaming เหมาะกับ real-time data transfer แต่ไม่เหมาะกับ long-term storage เพราะ: | ปัจจัย | WebSocket (เดิม) | Parquet Data Lake (ใหม่) | |--------|------------------|-------------------------| | **รูปแบบข้อมูล** | Binary/JSON streaming | Columnar format | | **Compression** | None หรือ basic | Snappy/Zstd 70-80% | | **Query capability** | Sequential scan only | Predicate pushdown | | **Storage cost/GB** | $0.023 (S3 Standard) | $0.005 (S3 Glacier + intelligent tiering) | | **Read cost** | High (full transfer) | Low (column projection) | | **Schema evolution** | ยากมาก | รองรับ built-in |

สถานการณ์ที่พบบ่อย: E-commerce AI Customer Service

สมมติคุณมีระบบ AI customer service สำหรับ e-commerce ที่: - รับ 50,000 conversations/วัน - เฉลี่ย 15 messages/conversation - เก็บข้อมูล 90 วันย้อนหลัง - ต้อง query หา "ผู้ใช้ที่ถามเรื่อง refund ในเดือนที่แล้ว" การใช้ WebSocket หมายความว่า คุณต้อง transfer ข้อมูลทั้งหมด แล้ว filter ที่ application layer — เปลือง bandwidth และเวลา

Parquet: คำตอบสำหรับ Analytical Workloads

ทำไม Parquet ถึงเหมาะสม

Parquet เป็น columnar storage format ที่ออกแบบมาสำหรับ analytical queries โดยเฉพาะ:

# ตัวอย่าง: เปรียบเทียบขนาดข้อมูล
import pyarrow as pa
import pyarrow.parquet as pq

JSON แบบ WebSocket ทั่วไป (1 conversation)
json_data = {
    "session_id": "sess_12345",
    "user_id": "user_67890",
    "messages": [
        {"role": "user", "content": "ต้องการคืนสินค้า", "timestamp": "2026-04-15T10:30:00Z"},
        {"role": "assistant", "content": "ช่วยบอกเลข order ได้ไหมครับ", "timestamp": "2026-04-15T10:30:05Z"},
        # ... messages อื่นๆ
    ],
    "metadata": {
        "page_url": "/checkout",
        "device": "mobile",
        "country": "TH"
    }
}

ขนาด JSON string: ~2.5 KB
json_size = len(str(json_data))  # ~2,500 bytes

Parquet columnar storage: บีบอัดแต่ละ column แยก
เมื่อ query แค่ messages.content + timestamp + user_id = 320 bytes
parquet_effective_size = 320  # bytes
compression_ratio = json_size / parquet_effective_size  # ~7.8x

Schema Design ที่ดี

สำหรับ L2 snapshot ที่ efficient ควรออกแบบ schema แบบนี้:

import pyarrow as pa
from pyarrow import parquet as pq

Define schema ที่ optimized สำหรับ L2 snapshot
schema = pa.schema([
    # Partition keys (ใช้ filter partition pruning)
    ('snapshot_date', pa.date32()),          # Partition by date
    ('hour_bucket', pa.uint8()),             # Partition by hour (0-23)
    
    # Primary identifiers
    ('session_id', pa.string()),
    ('user_id', pa.string()),
    ('conversation_id', pa.string()),
    
    # Message content (columnar for compression)
    ('messages', pa.list_(pa.struct([
        ('role', pa.string()),                # 'user'/'assistant'/'system'
        ('content', pa.string()),            # Actual text
        ('content_length', pa.uint32()),     # Stats for query optimization
        ('timestamp', pa.timestamp('ms')),
        ('model', pa.string()),              # Which model responded
        ('tokens_used', pa.uint32()),        # For cost tracking
        ('latency_ms', pa.float32())         # Performance monitoring
    ]))),
    
    # Session metadata (low cardinality - good for dictionary encoding)
    ('intent_class', pa.string()),           # Dictionary encoded
    ('satisfaction_score', pa.int8()),       # -1 to 5 scale
    ('resolution_status', pa.string()),      # 'resolved'/'escalated'/'pending'
    
    # Context (compress well due to repetition)
    ('product_context', pa.string()),        # JSON string of cart state
    ('previous_intents', pa.list_(pa.string())),
    
    # Technical metadata
    ('embedding_version', pa.string()),      # For consistency
    ('rag_references', pa.list_(pa.string())),  # Source documents
    ('error_flags', pa.uint16())             # Bitmask for various flags
])

Write with optimal compression
parquet_writer = pq.ParquetWriter(
    's3://bucket/l2_snapshots/',
    schema,
    compression='zstd',           # Better than snappy for text
    use_dictionary=True,          # For low-cardinality columns
    write_statistics=['content_length', 'timestamp', 'tokens_used']
)

การย้ายข้อมูลจาก WebSocket สู่ Parquet

ขั้นตอนที่ 1: สร้าง Consumer สำหรับ WebSocket Stream

import asyncio
import json
import pyarrow as pa
from datetime import datetime, timedelta
from collections import deque

class TardisWebSocketConsumer:
    """
    Consumer ที่อ่าน WebSocket stream แล้ว batch เขียน Parquet
    ออกแบบมาสำหรับ L2 snapshot migration
    """
    
    def __init__(self, ws_url: str, batch_size: int = 1000, flush_interval: int = 300):
        self.ws_url = ws_url
        self.batch_size = batch_size
        self.flush_interval = flush_interval  # seconds
        self.buffer = deque(maxlen=batch_size)
        self.last_flush = datetime.utcnow()
        
        # Arrow record batch builder
        self.field_names = [
            'session_id', 'user_id', 'conversation_id', 'snapshot_date', 
            'hour_bucket', 'messages', 'intent_class', 'satisfaction_score',
            'resolution_status', 'product_context', 'embedding_version'
        ]
        self.arrays = {name: [] for name in self.field_names}
    
    async def connect_and_consume(self, parquet_writer):
        """Main consumption loop"""
        import websockets
        
        async with websockets.connect(self.ws_url) as ws:
            print(f"Connected to {self.ws_url}")
            
            while True:
                try:
                    message = await asyncio.wait_for(ws.recv(), timeout=30)
                    data = json.loads(message)
                    
                    # Transform WebSocket format → Parquet-ready dict
                    transformed = self._transform_snapshot(data)
                    self._add_to_buffer(transformed)
                    
                    # Check if should flush
                    if self._should_flush():
                        await self._flush_to_parquet(parquet_writer)
                        
                except asyncio.TimeoutError:
                    # Send heartbeat
                    await ws.send(json.dumps({"type": "ping"}))
                    
                except Exception as e:
                    print(f"Error: {e}")
                    await asyncio.sleep(5)
    
    def _transform_snapshot(self, ws_data: dict) -> dict:
        """Transform WebSocket format to normalized schema"""
        dt = datetime.fromisoformat(ws_data['timestamp'].replace('Z', '+00:00'))
        
        return {
            'session_id': ws_data['session_id'],
            'user_id': ws_data.get('user_id', 'anonymous'),
            'conversation_id': ws_data['conversation_id'],
            'snapshot_date': dt.date(),
            'hour_bucket': dt.hour,
            'messages': [
                {
                    'role': msg['role'],
                    'content': msg['content'],
                    'content_length': len(msg['content']),
                    'timestamp': msg['timestamp'],
                    'model': msg.get('model', 'unknown'),
                    'tokens_used': msg.get('usage', {}).get('total_tokens', 0),
                    'latency_ms': msg.get('latency_ms', 0)
                }
                for msg in ws_data.get('messages', [])
            ],
            'intent_class': ws_data.get('metadata', {}).get('intent', 'unknown'),
            'satisfaction_score': ws_data.get('feedback', {}).get('score', -1),
            'resolution_status': ws_data.get('resolution', {}).get('status', 'pending'),
            'product_context': json.dumps(ws_data.get('cart_state', {})),
            'embedding_version': ws_data.get('embedding_config', {}).get('version', 'v1')
        }
    
    def _add_to_buffer(self, record: dict):
        """Add record to arrays"""
        for field in self.field_names:
            self.arrays[field].append(record.get(field))
    
    def _should_flush(self) -> bool:
        """Check if should flush based on size or time"""
        now = datetime.utcnow()
        size_full = len(self.arrays[self.field_names[0]]) >= self.batch_size
        time_elapsed = (now - self.last_flush).total_seconds() >= self.flush_interval
        return size_full or time_elapsed
    
    async def _flush_to_parquet(self, writer):
        """Write current buffer to Parquet"""
        import pyarrow as pa
        
        # Build record batch
        arrays = [
            pa.array(self.arrays[field], from_pandas=True) 
            for field in self.field_names
        ]
        batch = pa.RecordBatch.from_arrays(arrays, names=self.field_names)
        
        # Write
        writer.write_batch(batch)
        
        # Reset buffer
        for field in self.field_names:
            self.arrays[field] = []
        self.last_flush = datetime.utcnow()
        
        print(f"Flushed {len(batch)} records at {self.last_flush}")


Usage example
async def main():
    import pyarrow.parquet as pq
    
    writer = pq.ParquetWriter(
        's3://my-bucket/tardis-l2/year=2026/month=04/',
        schema,
        compression='zstd'
    )
    
    consumer = TardisWebSocketConsumer(
        ws_url="wss://api.example.com/tardis/stream",
        batch_size=5000,
        flush_interval=60
    )
    
    await consumer.connect_and_consume(writer)
    writer.close()

asyncio.run(main())

ขั้นตอนที่ 2: สร้าง Partition Strategy ที่ efficient

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, hour, month, year
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

def create_partitioned_dataset(input_path: str, output_path: str):
    """
    Repartition existing WebSocket data ไปเป็น Parquet พร้อม partitioning
    ที่เหมาะกับ query patterns ของ L2 snapshot
    """
    
    spark = SparkSession.builder \
        .appName("TardisL2Migration") \
        .config("spark.sql.adaptive.enabled", "true") \
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
        .config("spark.sql.parquet.compression.codec", "zstd") \
        .getOrCreate()
    
    # Read raw WebSocket JSON logs
    df = spark.read.json(input_path)
    
    # Parse timestamp
    df = df.withColumn('ts', col('timestamp').cast(TimestampType()))
    df = df.withColumn('dt', to_date(col('ts')))
    df = df.withColumn('hr', hour(col('ts')))
    df = df.withColumn('yr', year(col('ts')))
    df = df.withColumn('mo', month(col('ts')))
    
    # Add derived columns for better query performance
    df = df.withColumn('day_of_week', col('dt').cast('string'))
    df = df.withColumn('is_weekend', 
        col('day_of_week').isin(['Saturday', 'Sunday']).cast('int'))
    
    # Explode messages for granular analysis
    # (Keep both exploded AND nested for flexibility)
    df_messages = df.select(
        col('session_id'),
        col('user_id'),
        col('dt'),
        col('hr'),
        col('yr'),
        col('mo'),
        col('metadata.intent_class'),
        col('resolution.status'),
        col('feedback.score'),
        col('messages')
    )
    
    # Write with Hive-style partitioning
    # Partition by: year/month/day/hour - เหมาะกับ time-series queries
    df_messages.write \
        .mode('append') \
        .format('parquet') \
        .partitionBy('yr', 'mo', 'dt', 'hr') \
        .option('compression', 'zstd') \
        .save(output_path)
    
    # Create Delta Lake table สำหรับ ACID compliance
    spark.sql(f"""
        CREATE TABLE IF NOT EXISTS tardis_l2_snapshots
        USING DELTA
        LOCATION '{output_path}'
        PARTITIONED BY (yr, mo, dt, hr)
    """)
    
    # Optimize for query performance
    spark.sql(f"""
        OPTIMIZE tardis_l2_snapshots ZORDER BY (session_id, user_id)
    """)
    
    spark.stop()
    
    print(f"Migration complete. Data written to {output_path}")
    return output_path

ขั้นตอนที่ 3: Query Examples ที่ใช้งานจริง

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count, sum as spark_sum
from pyspark.sql.types import IntegerType

class TardisQueryEngine:
    """
    Query engine สำหรับ L2 snapshot analysis
    ออกแบบมาให้ใช้งานง่ายและ efficient
    """
    
    def __init__(self, data_path: str):
        self.spark = SparkSession.builder \
            .appName("TardisL2Query") \
            .config("spark.sql.parquet.enableVectorizedReader", "true") \
            .getOrCreate()
        self.data_path = data_path
        
    def get_hourly_conversation_volume(self, start_date: str, end_date: str):
        """นับจำนวน conversations ตามชั่วโมง"""
        df = self.spark.read.parquet(
            f"{self.data_path}/yr=*/mo=*/dt=*/hr=*",
            basePath=self.data_path
        ).filter(
            (col('dt') >= start_date) & (col('dt') <= end_date)
        )
        
        return df.groupBy('hr').agg(
            count('session_id').alias('total_sessions'),
            count(col('user_id')).alias('unique_users')
        ).orderBy('hr').collect()
    
    def find_refund_related_conversations(self, start_date: str, end_date: str):
        """หา conversations ที่เกี่ยวกับ refund"""
        df = self.spark.read.parquet(self.data_path).filter(
            col('dt').between(start_date, end_date)
        )
        
        # Filter by intent class
        refund_df = df.filter(
            col('intent_class').rlike('refund|return|เงิน|คืน')
        )
        
        return refund_df.select(
            'session_id',
            'user_id',
            'dt',
            'hr',
            'intent_class',
            'resolution.status',
            'feedback.score'
        ).toPandas()
    
    def calculate_monthly_ai_costs(self, year: int, month: int):
        """คำนวณค่าใช้จ่าย AI แยกตาม intent"""
        df = self.spark.read.parquet(
            f"{self.data_path}/yr={year}/mo={month}"
        )
        
        # Explode messages to get per-message stats
        exploded = df.select(
            col('session_id'),
            col('intent_class'),
            col('messages')
        ).withColumn('msg', col('messages')).drop('messages')
        
        return exploded.groupBy('intent_class').agg(
            count('session_id').alias('total_conversations'),
            avg(col('msg.tokens_used')).alias('avg_tokens_per_msg'),
            spark_sum(col('msg.tokens_used')).alias('total_tokens'),
            # ประมาณค่าใช้จ่าย (ใช้ HolySheep pricing)
            # DeepSeek V3.2: $0.42/MTok, เริ่มจาก embedding ก่อน
            (spark_sum(col('msg.tokens_used')) / 1_000_000 * 0.42).alias('estimated_cost_usd')
        ).orderBy(col('total_tokens').desc()).collect()
    
    def get_rag_performance_metrics(self, start_date: str, end_date: str):
        """วิเคราะห์ RAG effectiveness"""
        df = self.spark.read.parquet(self.data_path).filter(
            col('dt').between(start_date, end_date)
        )
        
        # Flatten message-level metrics
        from pyspark.sql.functions import explode
        
        msg_df = df.select(
            'session_id',
            explode('messages').alias('msg')
        ).select(
            col('session_id'),
            col('msg.model'),
            col('msg.latency_ms'),
            col('msg.tokens_used')
        )
        
        return msg_df.groupBy('model').agg(
            count('*').alias('message_count'),
            avg('latency_ms').alias('avg_latency_ms'),
            avg('tokens_used').alias('avg_tokens'),
            spark_sum('tokens_used').alias('total_tokens')
        ).toPandas()
    
    def close(self):
        self.spark.stop()


Usage with HolySheep for cost optimization
def analyze_and_optimize_costs(data_path: str):
    """
    ใช้ HolySheep API เพื่อ generate cost analysis report
    """
    engine = TardisQueryEngine(data_path)
    
    # Get monthly costs
    costs = engine.calculate_monthly_ai_costs(2026, 4)
    
    # Convert to report
    print("=== AI Costs Analysis (April 2026) ===")
    total_cost = 0
    for row in costs:
        print(f"Intent: {row.intent_class}")
        print(f"  Conversations: {row.total_conversations:,}")
        print(f"  Total Tokens: {row.total_tokens:,}")
        print(f"  Est. Cost: ${row.estimated_cost_usd:.2f}")
        print()
        total_cost += row.estimated_cost_usd
    
    print(f"Total Estimated Cost: ${total_cost:.2f}")
    
    # ถ้าใช้ HolySheep: ลด 85%+ 
    # HolySheep DeepSeek V3.2: $0.42/MTok (vs OpenAI $8/MTok)
    holysheep_cost = total_cost * 0.42 / 8  # 85%+ reduction
    print(f"Cost with HolySheep: ${holysheep_cost:.2f}")
    print(f"Savings: ${total_cost - holysheep_cost:.2f} ({(1 - holysheep_cost/total_cost)*100:.1f}%)")
    
    engine.close()

การใช้งานจริงกับ RAG System

Integration กับ Vector Database

from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
import pyarrow.parquet as pq
import numpy as np

class TardisRAGIntegration:
    """
    ดึงข้อมูลจาก Parquet L2 snapshots ไปสร้าง RAG knowledge base
    """
    
    def __init__(self, parquet_path: str, qdrant_url: str, collection_name: str):
        self.parquet_path = parquet_path
        self.qdrant = QdrantClient(url=qdrant_url)
        self.collection_name = collection_name
        
        # Initialize embedding model (ใช้ local model หรือ HolySheep)
        self.embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
        
    def ingest_conversations_to_vector_db(
        self, 
        start_date: str, 
        end_date: str,
        intent_filter: list = None,
        batch_size: int = 100
    ):
        """
        Ingest conversations ไปยัง vector database สำหรับ RAG
        """
        # Read from Parquet with partition pruning
        pf = pq.ParquetFile(self.parquet_path)
        
        # Use row groups for efficient reading
        table = pf.read(
            filters=[
                ('dt', '>=', start_date),
                ('dt', '<=', end_date)
            ]
        ).to_pandas()
        
        if intent_filter:
            table = table[table['intent_class'].isin(intent_filter)]
        
        print(f"Processing {len(table)} conversations...")
        
        points = []
        for idx, row in table.iterrows():
            # Extract conversation context
            messages = row['messages']
            conversation_text = "\n".join([
                f"{msg['role']}: {msg['content']}"
                for msg in messages
            ])
            
            # Create embedding
            embedding = self.embedding_model.encode(conversation_text)
            
            # Prepare payload
            payload = {
                "session_id": row['session_id'],
                "user_id": row['user_id'],
                "intent_class": row['intent_class'],
                "resolution_status": row['resolution_status'],
                "satisfaction_score": row['satisfaction_score'],
                "conversation_preview": conversation_text[:500],  # First 500 chars
                "message_count": len(messages),
                "date": str(row['dt'])
            }
            
            points.append({
                "id": idx,
                "vector": embedding.tolist(),
                "payload": payload
            })
            
            if len(points) >= batch_size:
                self.qdrant.upsert(
                    collection_name=self.collection_name,
                    points=points
                )
                points = []
        
        # Upload remaining
        if points:
            self.qdrant.upsert(
                collection_name=self.collection_name,
                points=points
            )
        
        print(f"Ingestion complete: {len(table)} conversations indexed")
        
    def query_rag(self, query: str, top_k: int = 5):
        """
        Query RAG system สำหรับ relevant past conversations
        """
        # Create query embedding
        query_embedding = self.embedding_model.encode(query)
        
        # Search
        results = self.qdrant.search(
            collection_name=self.collection_name,
            query_vector=query_embedding.tolist(),
            limit=top_k
        )
        
        return [
            {
                "session_id": r.payload['session_id'],
                "intent_class": r.payload['intent_class'],
                "preview": r.payload['conversation_preview'],
                "score": r.score
            }
            for r in results
        ]

เหมาะกับใคร / ไม่เหมาะกับใคร

✓ เหมาะกับใคร

- **ทีม Data Engineering** ที่รันระบบ Tardis หรือ similar conversation logging systems - **องค์กรที่มี AI customer service** ต้องการวิเคราะห์ historical conversations เป็นประจำ - **ทีมที่ใช้ RAG** และต้องการ indexing จาก conversation history - **Startup ที่ต้องการลดค่าใช้จ่ายด้าน storage** โดยไม่สูญเสีย query capability - **ทีมที่ต้องการ debug** หรือ replay conversation sessions

✗ ไม่เหมาะกับใคร

- **ระบบที่ต้องการ sub-second latency** สำหรับทุก query — Parquet ไม่เหมาะกับ real-time single-row lookups - **ข้อมูลขนาดเล็กมาก** (< 1GB) ที่ยังไม่คุ้มค่ากับการ setup infrastructure - **ทีมที่ไม่มี Spark หรือ query engine** — อาจใช้ DuckDB แทนได้แต่ต้องปรับ design

ราคาและ ROI

เปรียบเทียบค่าใช้จ่าย (ต่อเดือน, 100GB ข้อมูล)

| โซลูชัน | Storage Cost | Query Cost | Analytics Capability | Setup Effort | |---------|-------------|------------|---------------------|--------------| | **WebSocket + S3 Standard** | $2.30 | $0.00 | ต่ำ | ต่ำ | | **Raw JSON on S3** | $2.30 | $5-50 | ปานกลาง | ต่ำ | | **Parquet on S3 + Spark** | $0.50 | $2-15 | สูงมาก | ปานกลาง | | **Parquet + DuckDB** | $0.50 | $0-5 | สูง | ต่ำ | | **Delta Lake on Databricks** | $0.50 | $15-50 | สูงสุด | สูง |

การคำนวณ ROI

สมมติคุณมี **500GB ข้อมูล L2 snapshots** ต่อเดือน: **ก่อน (WebSocket):** - S3 Standard: 500GB × $0.023 = **$11.50/เดือน** - Egress + Query: ~$50-200/เดือน - **รวม: ~$61-211/เดือน** **หลัง (Parquet):** - S3 Intelligent Tiering: 500GB × $0.004 = **$2.00/เดือน** - Athena Query: ~$5-20/เดือน - **รวม: ~$7-22/เดื