Parquet Format Cost Optimization ลดค่าใช้จ่าย

Parquet Format คืออะไร

Apache Parquet เป็น Open-source Columnar Storage Format ที่ออกแบบมาสำหรับ Big Data Analytics จุดเด่นคือเก็บข้อมูลเป็นคอลัมน์ (Column-oriented) แทนที่จะเก็บเป็นแถว (Row-oriented) แบบ CSV ทำให้ Query ที่อ่านเฉพาะบางคอลัมน์ทำงานเร็วขึ้นหลายเท่า และบีบอัดข้อมูลได้ดีกว่ามากเพราะข้อมูลในคอลัมน์เดียวกันมักมีลักษณะคล้ายกัน

Parquet ถูกใช้อย่างแพร่หลายใน Data Lake, Data Warehouse และ Big Data Pipeline ทั้งบน Cloud (AWS S3 + Athena, GCS + BigQuery, Azure + Synapse) และ On-premise (Hadoop, Spark) เป็น Format มาตรฐานที่ทุก Data Tool รองรับ

เปรียบเทียบ Parquet กับ CSV และ JSON

การแปลง CSV เป็น Parquet ด้วย Python

# แปลง CSV เป็น Parquet ด้วย Pandas และ PyArrow
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os
import time

def csv_to_parquet(csv_path, parquet_path, compression="snappy",
                   partition_cols=None, row_group_size=100000):
    """
    แปลง CSV เป็น Parquet พร้อม Optimization
    compression: snappy (เร็ว), gzip (เล็กกว่า), zstd (สมดุล)
    partition_cols: คอลัมน์สำหรับ Partition เช่น ["year", "month"]
    row_group_size: จำนวนแถวต่อ Row Group (ยิ่งใหญ่ยิ่ง Compress ดี)
    """
    print(f"อ่าน CSV: {csv_path}")
    start = time.time()

    # อ่าน CSV
    df = pd.read_csv(csv_path, low_memory=False)
    csv_size = os.path.getsize(csv_path)

    # Optimize Data Types ก่อนเขียน Parquet
    for col in df.select_dtypes(include=["int64"]).columns:
        if df[col].min() >= 0 and df[col].max() < 2**31:
            df[col] = df[col].astype("int32")

    for col in df.select_dtypes(include=["float64"]).columns:
        df[col] = df[col].astype("float32")

    for col in df.select_dtypes(include=["object"]).columns:
        if df[col].nunique() / len(df) < 0.5:  # ถ้ามี Unique น้อยกว่า 50%
            df[col] = df[col].astype("category")

    # เขียน Parquet
    if partition_cols:
        table = pa.Table.from_pandas(df)
        pq.write_to_dataset(
            table, parquet_path,
            partition_cols=partition_cols,
            compression=compression,
        )
    else:
        df.to_parquet(
            parquet_path,
            engine="pyarrow",
            compression=compression,
            row_group_size=row_group_size,
            index=False,
        )

    parquet_size = os.path.getsize(parquet_path) if not partition_cols else \
        sum(os.path.getsize(os.path.join(dp, f))
            for dp, dn, filenames in os.walk(parquet_path)
            for f in filenames)

    elapsed = time.time() - start
    ratio = (1 - parquet_size / csv_size) * 100

    print(f"CSV Size:     {csv_size / 1024 / 1024:.1f} MB")
    print(f"Parquet Size: {parquet_size / 1024 / 1024:.1f} MB")
    print(f"Compression:  {ratio:.1f}% ลดลง")
    print(f"เวลา:         {elapsed:.1f} วินาที")

    return {"csv_mb": csv_size/1024/1024, "parquet_mb": parquet_size/1024/1024,
            "compression_pct": ratio}

# ตัวอย่างการใช้งาน
# csv_to_parquet("sales_data.csv", "sales_data.parquet", compression="zstd")
# csv_to_parquet("logs.csv", "logs_partitioned/",
#                partition_cols=["year", "month"], compression="snappy")

Query Performance Optimization ด้วย DuckDB

# เปรียบเทียบ Query Performance: CSV vs Parquet ด้วย DuckDB
import duckdb
import time

def benchmark_query(description, query, con):
    """วัดเวลา Query"""
    start = time.time()
    result = con.execute(query).fetchdf()
    elapsed = time.time() - start
    print(f"{description}: {elapsed:.3f}s ({len(result)} rows)")
    return elapsed

con = duckdb.connect()

# สร้างข้อมูลทดสอบ 10 ล้านแถว
con.execute("""
    CREATE TABLE sales AS
    SELECT
        i AS id,
        '2025-' || LPAD(CAST(1 + (i % 12) AS VARCHAR), 2, '0') || '-01' AS date,
        CASE i % 5
            WHEN 0 THEN 'Electronics'
            WHEN 1 THEN 'Clothing'
            WHEN 2 THEN 'Food'
            WHEN 3 THEN 'Books'
            ELSE 'Home'
        END AS category,
        ROUND(RANDOM() * 10000, 2) AS amount,
        'Customer_' || (i % 100000) AS customer_id,
        CASE i % 3
            WHEN 0 THEN 'Bangkok'
            WHEN 1 THEN 'Chiang Mai'
            ELSE 'Phuket'
        END AS region
    FROM generate_series(1, 10000000) t(i)
""")

# Export เป็น CSV และ Parquet
con.execute("COPY sales TO 'sales.csv' (HEADER)")
con.execute("COPY sales TO 'sales.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")

# Benchmark 1: Full Table Scan
t1 = benchmark_query("CSV Full Scan",
    "SELECT COUNT(*), SUM(amount) FROM 'sales.csv'", con)
t2 = benchmark_query("Parquet Full Scan",
    "SELECT COUNT(*), SUM(amount) FROM 'sales.parquet'", con)
print(f"  → Parquet เร็วกว่า {t1/t2:.1f}x\n")

# Benchmark 2: Column Pruning (อ่านแค่ 2 คอลัมน์)
t1 = benchmark_query("CSV 2 Columns",
    "SELECT category, SUM(amount) FROM 'sales.csv' GROUP BY category", con)
t2 = benchmark_query("Parquet 2 Columns",
    "SELECT category, SUM(amount) FROM 'sales.parquet' GROUP BY category", con)
print(f"  → Parquet เร็วกว่า {t1/t2:.1f}x\n")

# Benchmark 3: Predicate Pushdown (Filter ก่อนอ่าน)
t1 = benchmark_query("CSV with Filter",
    "SELECT * FROM 'sales.csv' WHERE category = 'Electronics' AND amount > 5000", con)
t2 = benchmark_query("Parquet with Filter",
    "SELECT * FROM 'sales.parquet' WHERE category = 'Electronics' AND amount > 5000", con)
print(f"  → Parquet เร็วกว่า {t1/t2:.1f}x")

Cost Optimization บน Cloud

# คำนวณ Cost Savings เมื่อเปลี่ยนจาก CSV เป็น Parquet บน AWS
import json

def calculate_cloud_savings(
    csv_size_gb,
    parquet_compression_ratio=0.85,  # 85% compression
    queries_per_month=1000,
    avg_columns_queried_pct=0.2,  # อ่านเฉลี่ย 20% ของคอลัมน์
    s3_price_per_gb=0.023,  # USD/GB/month
    athena_price_per_tb_scanned=5.0,  # USD/TB
):
    """คำนวณค่าใช้จ่ายที่ประหยัดได้"""
    parquet_size_gb = csv_size_gb * (1 - parquet_compression_ratio)

    # Storage Cost
    csv_storage_cost = csv_size_gb * s3_price_per_gb
    parquet_storage_cost = parquet_size_gb * s3_price_per_gb
    storage_savings = csv_storage_cost - parquet_storage_cost

    # Query Cost (Athena/BigQuery คิดตามข้อมูลที่ Scan)
    csv_scan_per_query_gb = csv_size_gb  # CSV ต้อง Scan ทั้งไฟล์
    parquet_scan_per_query_gb = parquet_size_gb * avg_columns_queried_pct

    csv_query_cost = (csv_scan_per_query_gb / 1024 * athena_price_per_tb_scanned
                      * queries_per_month)
    parquet_query_cost = (parquet_scan_per_query_gb / 1024 * athena_price_per_tb_scanned
                          * queries_per_month)
    query_savings = csv_query_cost - parquet_query_cost

    total_savings = storage_savings + query_savings
    total_savings_pct = total_savings / (csv_storage_cost + csv_query_cost) * 100

    return {
        "csv_storage_monthly": f"",
        "parquet_storage_monthly": f"",
        "storage_savings": f"/month",
        "csv_query_monthly": f"",
        "parquet_query_monthly": f"",
        "query_savings": f"/month",
        "total_savings": f"/month",
        "total_savings_pct": f"{total_savings_pct:.1f}%",
        "annual_savings": f"/year",
    }

# ตัวอย่าง: ข้อมูล 500 GB, Query 1000 ครั้ง/เดือน
result = calculate_cloud_savings(csv_size_gb=500, queries_per_month=1000)
print("=== Cost Savings: CSV → Parquet ===")
for k, v in result.items():
    print(f"  {k}: {v}")

Partitioning Strategy

• Partition by Date: เหมาะกับ Time-series Data เช่น Logs, Events, Transactions ใช้ year/month/day เป็น Partition Key
• Partition by Category: เหมาะเมื่อ Query Filter ด้วย Category เสมอ เช่น region, country, product_type
• ระวัง Over-partitioning: ถ้า Partition มีไฟล์เล็กมากๆ (Small Files Problem) จะช้ากว่าเพราะต้อง Open ไฟล์เยอะ แต่ละ Partition ควรมีข้อมูลอย่างน้อย 128 MB
• ใช้ Hive-style Partitioning: เก็บเป็น path/year=2026/month=03/data.parquet เพื่อให้ Query Engine ทำ Partition Pruning ได้

Best Practices สำหรับ Parquet

• เลือก Compression ที่เหมาะสม: snappy สำหรับความเร็ว, zstd สำหรับสมดุลระหว่างขนาดและความเร็ว, gzip สำหรับขนาดเล็กที่สุด
• ตั้ง Row Group Size ให้เหมาะสม: 128 MB - 1 GB ต่อ Row Group ใหญ่เกินไปจะใช้ Memory เยอะ เล็กเกินไปจะ Compress ไม่ดี
• Optimize Data Types: ใช้ int32 แทน int64 เมื่อเป็นไปได้ ใช้ Dictionary Encoding สำหรับ Low-cardinality Columns
• Sort Data ก่อนเขียน: Sort ตามคอลัมน์ที่ใช้ Filter บ่อยจะช่วยให้ Min/Max Statistics ทำงานดีขึ้น
• ใช้ Predicate Pushdown: เขียน WHERE clause ให้ตรงกับ Partition Key และ Sort Column เพื่อข้าม Row Groups ที่ไม่เกี่ยวข้อง

Parquet Format คืออะไรและต่างจาก CSV อย่างไร

Parquet เป็น Columnar Storage Format เก็บข้อมูลเป็นคอลัมน์แทนแถว ต่างจาก CSV ที่เก็บเป็นแถว ข้อดีคือบีบอัดได้ดีกว่า 60-90% อ่านเฉพาะคอลัมน์ที่ต้องการได้ มี Schema ในตัว และ Query เร็วกว่ามากสำหรับ Analytical Workload

สรุปและแนวทางปฏิบัติ

Parquet เป็น Format มาตรฐานสำหรับ Big Data Analytics ที่ช่วยลดค่าใช้จ่ายทั้ง Storage และ Query Cost บน Cloud ได้อย่างมีนัยสำคัญ การเปลี่ยนจาก CSV เป็น Parquet เป็นหนึ่งใน Quick Win ที่ง่ายที่สุดสำหรับ Cost Optimization สิ่งสำคัญคือเลือก Compression ที่เหมาะสม ตั้ง Partition Strategy ตาม Query Pattern ใช้ Data Type ที่เล็กที่สุดเท่าที่เป็นไปได้ และ Sort Data ก่อนเขียนเพื่อให้ Predicate Pushdown ทำงานดี