استراتژیهای پارتیشنبندی و بهینهسازی ذخیرهسازی در Data Lake (قالب Parquet و فشردهسازی ZSTD)

کالبدشکافی ساختار درونی فایلهای Apache Parquet

فرمت Apache Parquet ستون فقرات دریاچههای داده مدرن (Data Lakes) و فرمتهای جدولی مثل Apache Iceberg و Delta Lake است. برخلاف فایلهای CSV یا JSON که دادهها را سطر به سطر ذخیره میکنند، Parquet از یک معماری ترکیبی هوشمند به نام PAX (Partition Attributes Across) استفاده میکند که ابتدا جدول را به صورت افقی و سپس به صورت ستونی تقسیمبندی مینماید:

• ۱. لایه Row Group (گروههای سطری): هر فایل Parquet به چند بلوک افقی بزرگ (معمولاً بین ۶۴ تا ۲۵۶ مگابایت یا حدود ۱۰۰ هزار تا ۱ میلیون سطر) تقسیم میشود.
• ۲. لایه Column Chunk: درون هر Row Group، مقادیر هر ستون به صورت کاملاً مجزا و پشت سر هم در یک Column Chunk ذخیره میشوند.
• ۳. لایه Data Page: هر Column Chunk به صفحات کوچکتر (معمولاً ۱ مگابایت) شکسته میشود که کدگذاریهایی مانند Dictionary Encoding و RLE (Run-Length Encoding) به همراه فشردهسازی روی آنها اعمال میگردد.
• ۴. بخش حیاتی File Footer: در انتهای فایل Parquet، متادیتای کامل اسکیما و آمار دقیق هر Column Chunk در هر Row Group (شامل مقادیر min، max و null_count) نگهداری میشود.

رمزگشایی از Predicate Pushdown: چرا مرتبسازی دادهها پیش از نوشتن حیاتی است؟

وقتی موتور کوئری (مانند DuckDB، Trino یا Spark) میخواهد کوئری SELECT * FROM events WHERE user_id = 84920 را اجرا کند، ابتدا فقط چند کیلوبایت انتهایی فایل (Footer) را از روی S3/MinIO میخواند. سپس آمار min و max ستون user_id را در هر Row Group بررسی میکند. اگر عدد 84920 خارج از بازه [min, max] یک Row Group باشد، موتور کوئری کل آن بلوک صد مگابایتی را بدون دانلود کردن رد میکند (Row Group Skipping).

اما یک دام بزرگ مهندسی وجود دارد: اگر دادهها پیش از نوشته شدن در فایل Parquet بر اساس ستونهای پرکاربرد در شرط WHERE مرتب (Sort / Z-Order) نشده باشند، مقادیر کوچک و بزرگ به صورت تصادفی در تمام Row Groupها پخش میشوند. در نتیجه، بازه [min, max] در تمام Row Groupها تقریباً یکسان (مثلاً از 1 تا 9999999) خواهد بود و مکانیسم Predicate Pushdown به طور کامل از کار میافتد!

چرا ZSTD برنده مطلق در برابر Snappy و Gzip است؟

سالها الگوریتم Snappy پیشفرض اکوسیستم Hadoop بود، اما در سختافزارهای مدرن، الگوریتم ZSTD (Zstandard) sviluppato توسط متا، استاندارد طلایی محسوب میشود. فشردهسازی ZSTD در سطح level=3 حجم فایلهای Parquet را نسبت به Snappy تا ۳۵ الی ۴۵ درصد کوچکتر میکند (کاهش مستقیم هزینه ذخیرهسازی و پهنای باند شبکه S3)، در حالی که سرعت رمزگشایی (Decompression) آن تقریباً همتراز با Snappy و چندین برابر سریعتر از Gzip است.

معضل فاجعهبار Small Files Problem در Object Storage

در پایپلاینهای استریمینگ که هر دقیقه یک فایل Parquet روی S3 یا MinIO مینویسند، پس از یک ماه با صدها هزار فایل ۱ تا ۲ مگابایتی در هر پارتیشن مواجه میشوید. در Object Storageها، هر درخواست GET و HEAD دارای تاخیر شبکه (TTFB) حدود ۲۰ تا ۵۰ میلیثانیه است. خواندن ۱۰۰ هزار فایل کوچک ۱ مگابایتی (مجموعاً ۱۰۰ گیگابایت) میتواند ۲۰ دقیقه طول بکشد، در حالی که خواندن همان ۱۰۰ گیگابایت در قالب ۴۰۰ فایل ۲۵۰ مگابایتی کمتر از ۱۰ ثانیه زمان میبرد!

علاوه بر این، پارتیشنبندی بیش از حد به سبک Hive (مثلاً year=.../month=.../day=.../hour=.../country=...) یا همان Over-Partitioning، تعداد فایلهای کوچک را به صورت نمایی افزایش میدهد. قانون طلایی این است که حجم هر فایل فیزیکی Parquet باید بین ۱۲۸ تا ۵۱۲ مگابایت باشد.

پیادهسازی سرویس Compaction و بهینهسازی Parquet با PyArrow

اسکریپت زیر یک موتور Compaction (یکپارچهسازی) برای دریاچه داده است که هزاران فایل کوچک یک پارتیشن را به صورت استریم میخواند، آنها را بر اساس کلیدهای پرکاربرد مرتب میکند تا Predicate Pushdown و Dictionary Encoding به حداکثر کارایی برسند، و خروجی را با فشردهسازی ZSTD در فایلهای استاندارد ۲۵۶ مگابایتی بازنویسی مینماید:

from pathlib import Path
from typing import List
import pyarrow as pa
import pyarrow.dataset as ds
import pyarrow.parquet as pq

class DataLakeParquetCompactor:
    """
    Compacts fragmented small Parquet files into optimal Row-Group sized,
    sorted, ZSTD-compressed Parquet files for maximum Predicate Pushdown.
    """
    def __init__(
        self,
        target_rows_per_file: int = 2_500_000,
        row_group_size: int = 250_000,
        zstd_level: int = 3
    ):
        self.target_rows_per_file = target_rows_per_file
        self.row_group_size = row_group_size
        self.zstd_level = zstd_level

    def compact_partition(
        self,
        input_dir: str,
        output_dir: str,
        sort_keys: List[str]
    ) -> None:
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        dataset = ds.dataset(input_dir, format='parquet')

        # Read fragmented files into an Arrow Table and sort by high-selectivity keys
        table: pa.Table = dataset.to_table()
        sort_indices = pa.compute.sort_indices(
            table,
            sort_keys=[(key, 'ascending') for key in sort_keys]
        )
        sorted_table = table.take(sort_indices)

        parquet_format = ds.ParquetFileFormat()
        write_options = parquet_format.make_write_options(
            compression='zstd',
            compression_level=self.zstd_level,
            use_dictionary=True,
            write_statistics=True,
            data_page_version='2.0'
        )

        ds.write_dataset(
            data=sorted_table,
            base_dir=output_dir,
            format=parquet_format,
            file_options=write_options,
            max_rows_per_file=self.target_rows_per_file,
            min_rows_per_group=self.row_group_size,
            max_rows_per_group=self.row_group_size,
            existing_data_behavior='overwrite_or_ignore',
            basename_template='compacted-part-{i}.parquet'
        )

if __name__ == '__main__':
    compactor = DataLakeParquetCompactor()
    compactor.compact_partition(
        input_dir='/data/lake/raw/events/dt=2025-10-27',
        output_dir='/data/lake/optimized/events/dt=2025-10-27',
        sort_keys=['tenant_id', 'event_type', 'event_timestamp']
    )