معماری خطوط پردازش Zero-Copy با Apache Arrow، Polars و DuckDB (گذر از پانداس برای سرعت ۱۰ برابری)

import polars as pl
import duckdb
from pathlib import Path

class ZeroCopyAnalyticalPipeline:
    def __init__(self, max_memory_gb: int = 8, threads: int = 8, temp_dir: str = '/tmp/duckdb_spill'):
        Path(temp_dir).mkdir(parents=True, exist_ok=True)
        self.con = duckdb.connect(database=':memory:')
        # Configure DuckDB for Out-of-Core execution to prevent OOM on massive joins
        self.con.execute(f"SET memory_limit='{max_memory_gb}GB';")
        self.con.execute(f"SET threads={threads};")
        self.con.execute(f"SET temp_directory='{temp_dir}';")
        self.con.execute("SET preserve_insertion_order=false;")

    def run_etl(self, events_parquet_glob: str, users_parquet_path: str, output_path: str) -> None:
        # Step 1: Build Lazy Logical Plan in Polars (Predicate & Projection Pushdown)
        lazy_events = (
            pl.scan_parquet(events_parquet_glob)
            .filter(
                (pl.col('event_timestamp') >= pl.lit('2025-01-01').str.to_datetime()) &
                (pl.col('status_code') == 200) &
                (pl.col('revenue_usd').is_not_null())
            )
            .select([
                'user_id',
                'session_id',
                'event_type',
                'revenue_usd',
                'event_timestamp'
            ])
        )

        lazy_users = (
            pl.scan_parquet(users_parquet_path)
            .filter(pl.col('is_active') == True)
            .select(['user_id', 'country_code', 'acquisition_channel'])
        )

        # Step 2: Register Polars LazyFrames in DuckDB via Arrow C Stream Interface (Zero-Copy)
        self.con.register('filtered_events', lazy_events)
        self.con.register('active_users', lazy_users)

        # Step 3: Execute Vectorized Out-of-Core Window Aggregation & Direct Parquet Export
        query = f"""
        COPY (
            WITH user_session_metrics AS (
                SELECT
                    e.user_id,
                    u.country_code,
                    u.acquisition_channel,
                    e.session_id,
                    COUNT(*) AS events_in_session,
                    SUM(e.revenue_usd) AS session_revenue,
                    ROW_NUMBER() OVER (
                        PARTITION BY e.user_id
                        ORDER BY SUM(e.revenue_usd) DESC
                    ) AS session_rank
                FROM filtered_events e
                INNER JOIN active_users u ON e.user_id = u.user_id
                GROUP BY e.user_id, u.country_code, u.acquisition_channel, e.session_id
            )
            SELECT *
            FROM user_session_metrics
            WHERE session_rank <= 3
        ) TO '{output_path}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 131072);
        """
        self.con.execute(query)

if __name__ == '__main__':
    pipeline = ZeroCopyAnalyticalPipeline(max_memory_gb=6, threads=8)
    pipeline.run_etl(
        events_parquet_glob='s3://data-lake/raw/events/*.parquet',
        users_parquet_path='s3://data-lake/dim/users.parquet',
        output_path='/data/curated/top_user_sessions.parquet'
    )