Opinion: You Should Switch from Pandas 2.2 to Polars 1.0 for Python Data Work

import time
import pandas as pd
import polars as pl
from typing import Union, Dict, Any
import logging

# Configure logging for error handling
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def benchmark_pandas_processing(csv_path: str) -> Dict[str, Union[float, str]]:
    """Process CSV with Pandas 2.2, return timing and memory metrics."""
    result = {"tool": "pandas-2.2", "load_time_ms": 0, "process_time_ms": 0, "error": None}
    try:
        # Start load timer
        load_start = time.perf_counter()

        # Read CSV with error handling for parse issues
        try:
            df = pd.read_csv(
                csv_path,
                parse_dates=["transaction_date"],
                dtype={"user_id": "int64", "amount": "float64"},
                on_bad_lines="skip"  # Skip malformed lines
            )
        except FileNotFoundError:
            result["error"] = f"Pandas: File not found at {csv_path}"
            logger.error(result["error"])
            return result
        except pd.errors.ParserError as e:
            result["error"] = f"Pandas: Parse error: {str(e)}"
            logger.error(result["error"])
            return result

        load_end = time.perf_counter()
        result["load_time_ms"] = (load_end - load_start) * 1000

        # Start processing timer
        process_start = time.perf_counter()

        # Transformations: filter, groupby, aggregate
        processed = df[
            (df["amount"] > 10) & (df["transaction_date"] > "2024-01-01")
        ].groupby("user_id").agg(
            total_spend=("amount", "sum"),
            avg_spend=("amount", "mean"),
            transaction_count=("amount", "count")
        ).reset_index()

        process_end = time.perf_counter()
        result["process_time_ms"] = (process_end - process_start) * 1000

        # Log success
        logger.info(f"Pandas processed {len(df)} rows, output {len(processed)} user groups")
        return result

    except Exception as e:
        result["error"] = f"Pandas: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

def benchmark_polars_processing(csv_path: str) -> Dict[str, Union[float, str]]:
    """Process CSV with Polars 1.0, return timing and memory metrics."""
    result = {"tool": "polars-1.0", "load_time_ms": 0, "process_time_ms": 0, "error": None}
    try:
        # Start load timer
        load_start = time.perf_counter()

        # Read CSV with Polars' strict error handling
        try:
            df = pl.read_csv(
                csv_path,
                try_parse_dates=True,
                dtypes={"user_id": pl.Int64, "amount": pl.Float64},
                on_bad_lines="skip"  # Skip malformed lines
            )
        except FileNotFoundError:
            result["error"] = f"Polars: File not found at {csv_path}"
            logger.error(result["error"])
            return result
        except pl.exceptions.ComputeError as e:
            result["error"] = f"Polars: Compute error: {str(e)}"
            logger.error(result["error"])
            return result

        load_end = time.perf_counter()
        result["load_time_ms"] = (load_end - load_start) * 1000

        # Start processing timer
        process_start = time.perf_counter()

        # Transformations: filter, groupby, aggregate (lazy evaluation by default)
        processed = df.lazy().filter(
            (pl.col("amount") > 10) & (pl.col("transaction_date") > pl.date(2024, 1, 1))
        ).groupby("user_id").agg(
            total_spend=pl.col("amount").sum(),
            avg_spend=pl.col("amount").mean(),
            transaction_count=pl.col("amount").count()
        ).collect()  # Trigger execution

        process_end = time.perf_counter()
        result["process_time_ms"] = (process_end - process_start) * 1000

        # Log success
        logger.info(f"Polars processed {df.height} rows, output {processed.height} user groups")
        return result

    except Exception as e:
        result["error"] = f"Polars: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

if __name__ == "__main__":
    # Path to 1GB synthetic CSV (generate with scripts/fake_data_generator.py)
    CSV_PATH = "data/transactions_1gb.csv"

    # Run Pandas benchmark
    pandas_result = benchmark_pandas_processing(CSV_PATH)
    # Run Polars benchmark
    polars_result = benchmark_polars_processing(CSV_PATH)

    # Print comparison
    print("\n=== Benchmark Results ===")
    for res in [pandas_result, polars_result]:
        if res["error"]:
            print(f"{res['tool']}: ERROR - {res['error']}")
        else:
            print(f"{res['tool']}: Load {res['load_time_ms']:.2f}ms, Process {res['process_time_ms']:.2f}ms")

    # Calculate speedup if no errors
    if not pandas_result["error"] and not polars_result["error"]:
        total_pandas = pandas_result["load_time_ms"] + pandas_result["process_time_ms"]
        total_polars = polars_result["load_time_ms"] + polars_result["process_time_ms"]
        speedup = total_pandas / total_polars
        print(f"\nPolars 1.0 total speedup vs Pandas 2.2: {speedup:.2f}x")

import psutil
import time
import pandas as pd
import polars as pl
from typing import List, Dict, Any
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def get_memory_usage_mb() -> float:
    """Return current process memory usage in MB."""
    process = psutil.Process()
    return process.memory_info().rss / (1024 * 1024)

def pandas_streaming_process(parquet_dir: str, output_path: str) -> Dict[str, Any]:
    """Process 10GB+ Parquet dataset in chunks with Pandas 2.2."""
    result = {"tool": "pandas-2.2", "total_time_ms": 0, "peak_memory_mb": 0, "error": None}
    try:
        process_start = time.perf_counter()
        peak_memory = 0
        chunk_results = []

        # Read Parquet in chunks (Pandas 2.2 supports chunked reading for Parquet)
        try:
            parquet_file = pd.read_parquet(parquet_dir, chunksize=100_000)
        except FileNotFoundError:
            result["error"] = f"Pandas: Directory not found: {parquet_dir}"
            logger.error(result["error"])
            return result
        except Exception as e:
            result["error"] = f"Pandas: Parquet read error: {str(e)}"
            logger.error(result["error"])
            return result

        for chunk_num, chunk in enumerate(parquet_file):
            # Update peak memory
            current_mem = get_memory_usage_mb()
            peak_memory = max(peak_memory, current_mem)

            # Process chunk: filter, aggregate
            processed_chunk = chunk[chunk["price"] > 50].groupby("category").agg(
                avg_price=("price", "mean"),
                item_count=("price", "count")
            ).reset_index()
            chunk_results.append(processed_chunk)

            logger.info(f"Pandas: Processed chunk {chunk_num}, {len(chunk)} rows, current memory {current_mem:.2f}MB")

        # Combine chunk results
        final_df = pd.concat(chunk_results, ignore_index=True)
        # Write output
        final_df.to_csv(output_path, index=False)

        process_end = time.perf_counter()
        result["total_time_ms"] = (process_end - process_start) * 1000
        result["peak_memory_mb"] = peak_memory

        logger.info(f"Pandas: Final output {len(final_df)} rows, written to {output_path}")
        return result

    except Exception as e:
        result["error"] = f"Pandas: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

def polars_streaming_process(parquet_dir: str, output_path: str) -> Dict[str, Any]:
    """Process 10GB+ Parquet dataset with Polars 1.0 lazy scanning."""
    result = {"tool": "polars-1.0", "total_time_ms": 0, "peak_memory_mb": 0, "error": None}
    try:
        process_start = time.perf_counter()
        peak_memory = 0

        # Scan Parquet lazily (Polars doesn't load full dataset into memory)
        try:
            df_lazy = pl.scan_parquet(parquet_dir)
        except FileNotFoundError:
            result["error"] = f"Polars: Directory not found: {parquet_dir}"
            logger.error(result["error"])
            return result
        except pl.exceptions.ComputeError as e:
            result["error"] = f"Polars: Parquet scan error: {str(e)}"
            logger.error(result["error"])
            return result

        # Define processing pipeline (lazy, no execution yet)
        processed_lazy = df_lazy.filter(pl.col("price") > 50).groupby("category").agg(
            avg_price=pl.col("price").mean(),
            item_count=pl.col("price").count()
        )

        # Execute pipeline and write directly to CSV (no intermediate full dataframe)
        processed_lazy.sink_csv(output_path)

        # Update peak memory (Polars uses minimal memory for lazy scans)
        current_mem = get_memory_usage_mb()
        peak_memory = current_mem

        process_end = time.perf_counter()
        result["total_time_ms"] = (process_end - process_start) * 1000
        result["peak_memory_mb"] = peak_memory

        logger.info(f"Polars: Output written to {output_path}, peak memory {peak_memory:.2f}MB")
        return result

    except Exception as e:
        result["error"] = f"Polars: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

if __name__ == "__main__":
    PARQUET_DIR = "data/transactions_10gb/*.parquet"  # Glob pattern for multiple Parquet files
    OUTPUT_DIR = "output"

    # Run Pandas streaming benchmark
    pandas_result = pandas_streaming_process(PARQUET_DIR, f"{OUTPUT_DIR}/pandas_output.csv")
    # Run Polars streaming benchmark
    polars_result = polars_streaming_process(PARQUET_DIR, f"{OUTPUT_DIR}/polars_output.csv")

    # Print comparison
    print("\n=== Streaming Benchmark Results ===")
    for res in [pandas_result, polars_result]:
        if res["error"]:
            print(f"{res['tool']}: ERROR - {res['error']}")
        else:
            print(f"{res['tool']}: Total time {res['total_time_ms']:.2f}ms, Peak memory {res['peak_memory_mb']:.2f}MB")

    # Calculate memory reduction
    if not pandas_result["error"] and not polars_result["error"]:
        mem_reduction = ((pandas_result["peak_memory_mb"] - polars_result["peak_memory_mb"]) / pandas_result["peak_memory_mb"]) * 100
        print(f"\nPolars 1.0 memory reduction vs Pandas 2.2: {mem_reduction:.2f}%")

import time
import pandas as pd
import polars as pl
from datetime import datetime
from typing import Dict, Union
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def pandas_timeseries_resample(csv_path: str) -> Dict[str, Union[float, str]]:
    """Resample minute-level time series data to hourly with Pandas 2.2."""
    result = {"tool": "pandas-2.2", "load_time_ms": 0, "resample_time_ms": 0, "error": None}
    try:
        # Load data
        load_start = time.perf_counter()
        try:
            df = pd.read_csv(
                csv_path,
                parse_dates=["timestamp"],
                index_col="timestamp",
                dtype={"open": "float64", "high": "float64", "low": "float64", "close": "float64", "volume": "int64"}
            )
        except FileNotFoundError:
            result["error"] = f"Pandas: File not found: {csv_path}"
            logger.error(result["error"])
            return result
        except pd.errors.ParserError as e:
            result["error"] = f"Pandas: Parse error: {str(e)}"
            logger.error(result["error"])
            return result
        load_end = time.perf_counter()
        result["load_time_ms"] = (load_end - load_start) * 1000

        # Resample to hourly OHLCV
        resample_start = time.perf_counter()
        try:
            resampled = df.resample("H").agg({
                "open": "first",
                "high": "max",
                "low": "min",
                "close": "last",
                "volume": "sum"
            }).dropna()
        except Exception as e:
            result["error"] = f"Pandas: Resample error: {str(e)}"
            logger.error(result["error"])
            return result
        resample_end = time.perf_counter()
        result["resample_time_ms"] = (resample_end - resample_start) * 1000

        logger.info(f"Pandas: Resampled {len(df)} rows to {len(resampled)} hourly bars")
        return result

    except Exception as e:
        result["error"] = f"Pandas: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

def polars_timeseries_resample(csv_path: str) -> Dict[str, Union[float, str]]:
    """Resample minute-level time series data to hourly with Polars 1.0."""
    result = {"tool": "polars-1.0", "load_time_ms": 0, "resample_time_ms": 0, "error": None}
    try:
        # Load data
        load_start = time.perf_counter()
        try:
            df = pl.read_csv(
                csv_path,
                try_parse_dates=True,
                dtypes={
                    "open": pl.Float64,
                    "high": pl.Float64,
                    "low": pl.Float64,
                    "close": pl.Float64,
                    "volume": pl.Int64
                }
            )
        except FileNotFoundError:
            result["error"] = f"Polars: File not found: {csv_path}"
            logger.error(result["error"])
            return result
        except pl.exceptions.ComputeError as e:
            result["error"] = f"Polars: Compute error: {str(e)}"
            logger.error(result["error"])
            return result
        load_end = time.perf_counter()
        result["load_time_ms"] = (load_end - load_start) * 1000

        # Resample to hourly OHLCV (Polars uses groupby with time binning)
        resample_start = time.perf_counter()
        try:
            # Truncate timestamp to hourly
            resampled = df.with_columns(
                pl.col("timestamp").dt.truncate("1h").alias("hourly_timestamp")
            ).groupby("hourly_timestamp").agg(
                open=pl.col("open").first(),
                high=pl.col("high").max(),
                low=pl.col("low").min(),
                close=pl.col("close").last(),
                volume=pl.col("volume").sum()
            ).drop_nulls()
        except Exception as e:
            result["error"] = f"Polars: Resample error: {str(e)}"
            logger.error(result["error"])
            return result
        resample_end = time.perf_counter()
        result["resample_time_ms"] = (resample_end - resample_start) * 1000

        logger.info(f"Polars: Resampled {df.height} rows to {resampled.height} hourly bars")
        return result

    except Exception as e:
        result["error"] = f"Polars: Unexpected error: {str(e)}"
        logger.error(result["error"])
        return result

if __name__ == "__main__":
    CSV_PATH = "data/stock_minute_data.csv"  # 5 years of minute data: ~2.6M rows

    # Run benchmarks
    pandas_res = pandas_timeseries_resample(CSV_PATH)
    polars_res = polars_timeseries_resample(CSV_PATH)

    # Print results
    print("\n=== Time Series Resampling Benchmark ===")
    for res in [pandas_res, polars_res]:
        if res["error"]:
            print(f"{res['tool']}: ERROR - {res['error']}")
        else:
            print(f"{res['tool']}: Load {res['load_time_ms']:.2f}ms, Resample {res['resample_time_ms']:.2f}ms")

    # Calculate speedup
    if not pandas_res["error"] and not polars_res["error"]:
        total_pandas = pandas_res["load_time_ms"] + pandas_res["resample_time_ms"]
        total_polars = polars_res["load_time_ms"] + polars_res["resample_time_ms"]
        speedup = total_pandas / total_polars
        print(f"\nPolars 1.0 total speedup vs Pandas 2.2: {speedup:.2f}x")

# Bad: Naive loop concatenation (slow, high memory)
import polars as pl
dfs = []
for file in ["data/file1.csv", "data/file2.csv", ..., "data/file100.csv"]:
    dfs.append(pl.read_csv(file))
combined = pl.concat(dfs)  # Creates full copies of all dataframes

# Good: Lazy scan with glob pattern (fast, low memory)
combined = pl.scan_csv("data/file*.csv").collect()  # Scans files lazily, no intermediate copies

# Good: vstack for small numbers of dataframes
df1 = pl.read_csv("data/file1.csv")
df2 = pl.read_csv("data/file2.csv")
combined = df1.vstack(df2)  # Faster than pl.concat([df1, df2])

# Bad: Pandas silent type inference
import pandas as pd
df = pd.read_csv("data/transactions.csv")  # "amount" column with "N/A" values becomes object type
df["amount"] * 2  # No error, but returns NaN for invalid values silently

# Good: Polars strict type inference
import polars as pl
try:
    df = pl.read_csv("data/transactions.csv", dtypes={"amount": pl.Float64})
except pl.exceptions.ComputeError as e:
    print(f"Type error: {e}")  # Catches invalid values immediately

# Good: Lenient casting for malformed values
df = pl.read_csv("data/transactions.csv").with_columns(
    pl.col("amount").cast(pl.Float64, strict=False)  # Invalid values become null
)

# Bad: Pandas parallel I/O with multiprocessing (complex, high overhead)
import pandas as pd
from multiprocessing import Pool

def read_pandas(path):
    return pd.read_csv(path)

with Pool(4) as p:
    dfs = p.map(read_pandas, ["s3://bucket/file1.csv", ..., "s3://bucket/file4.csv"])

# Good: Polars native async I/O (simple, low overhead)
import polars as pl
import asyncio

async def read_polars_async(paths):
    tasks = [pl.read_csv_async(path) for path in paths]
    return await asyncio.gather(*tasks)

dfs = asyncio.run(read_polars_async(["s3://bucket/file1.csv", ..., "s3://bucket/file4.csv"]))