diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml new file mode 100644 index 0000000..1096b30 --- /dev/null +++ b/.github/workflows/benchmark.yml @@ -0,0 +1,142 @@ +name: Dataframe Benchmarking + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: # Allow manual trigger + schedule: + # Run weekly on Sundays at 06:00 UTC + - cron: '0 6 * * 0' + +jobs: + benchmark: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python with uv + uses: astral-sh/setup-uv@v3 + with: + enable-cache: true + cache-dependency-glob: "requirements.in" + + - name: Set up Python + run: uv python install + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y bc + + - name: Run benchmark suite + run: | + # Make run.sh executable + chmod +x run.sh + + # Run the benchmark suite + bash run.sh + env: + # Set environment variable to indicate GitHub Actions + GITHUB_ACTIONS: true + + - name: Display Summary Statistics + run: | + echo "::group::Summary Statistics" + if [ -f "outputs/summary_statistics.md" ]; then + cat outputs/summary_statistics.md + else + echo "Summary statistics file not found" + fi + echo "::endgroup::" + + - name: Prepare benchmark artifacts + run: | + # Create a comprehensive benchmark report + mkdir -p benchmark-results + + # Copy all output files + cp -r outputs/* benchmark-results/ 2>/dev/null || true + cp -r artifacts/* benchmark-results/ 2>/dev/null || true + + # Create a summary report + cat > benchmark-results/README.md << 'EOF' + # Dataframe Benchmarking Results + + This directory contains the complete results from the dataframe benchmarking suite. + + ## Files Description + + ### Performance Results + - `comparison_table.csv` - Main performance comparison across frameworks + - `summary_statistics.csv` - Statistical summary of benchmark results + - `summary_statistics.md` - Markdown formatted summary statistics + - `speedup_comparison.csv` - Relative performance metrics (if available) + - `cache_effectiveness.csv` - Cache performance analysis (if available) + + ### Data Verification + - `hash_comparison.csv` - Hash comparison for result verification + - `processed_raw_data.csv` - Raw benchmark data + + ### Detailed Results + - `results/` - Individual operation results by framework + + ### Profiling Data + - `pyinstrument/` - Detailed profiling reports (HTML format) + + ### Environment Info + - `requirements_*.txt` - Python environment details and package versions + + ## Frameworks Tested + + - **pandas**: Traditional dataframe library + - **fireducks**: High-performance pandas-compatible library + - **polars**: Fast dataframes with lazy evaluation + + ## Benchmark Operations + + The suite tests 23 different operations including: + - Basic filtering and aggregations + - Complex multi-table joins + - Window functions and rolling operations + - String and datetime operations + - Statistical computations + - Large data concatenation and sorting + + Generated on: $(date -u '+%Y-%m-%d %H:%M:%S UTC') + Commit: $GITHUB_SHA + Workflow: $GITHUB_RUN_ID + EOF + + - name: Upload benchmark results + uses: actions/upload-artifact@v4 + with: + name: benchmark-results-${{ github.run_id }} + path: benchmark-results/ + retention-days: 30 + compression-level: 6 + + - name: Upload profiling reports + uses: actions/upload-artifact@v4 + with: + name: profiling-reports-${{ github.run_id }} + path: outputs/pyinstrument/ + retention-days: 30 + compression-level: 6 + if: always() + + - name: Summary + run: | + echo "::notice title=Benchmark Complete::Benchmark suite completed successfully! Check the artifacts for detailed results." + + # Display summary statistics if available + if [ -f "outputs/summary_statistics.md" ]; then + echo "::group::Summary Statistics" + cat outputs/summary_statistics.md + echo "::endgroup::" + else + echo "Summary statistics file not found" + fi \ No newline at end of file diff --git a/.gitignore b/.gitignore index b7faf40..f9a4a6b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,8 @@ +# Generated files: +artifacts/ +data/ +outputs/ + # Byte-compiled / optimized / DLL files __pycache__/ *.py[codz] diff --git a/00_tools.py b/00_tools.py new file mode 100644 index 0000000..40f2d2e --- /dev/null +++ b/00_tools.py @@ -0,0 +1,77 @@ +import hashlib +import os +import time +from datetime import datetime +from typing import Any, Callable, Dict + + +def time_operation( + operation_name: str, + df_lib: Any, + func: Callable[..., Any], + *args: Any, + **kwargs: Any, +) -> Dict[str, Any]: + """Time a function and return the result and execution info""" + + framework = df_lib.__name__ + if framework == "fireducks.pandas": + framework = "fireducks" + + start_time = time.perf_counter() + result = func(*args, **kwargs) + # Force evaluation for lazy operations + if hasattr(result, "compute"): + result = result.compute() + elif hasattr(result, "values"): + _ = result.values # Access values to force computation + elif isinstance(result, df_lib.DataFrame): + _ = len(result) # Force evaluation by accessing length + end_time = time.perf_counter() + execution_time = end_time - start_time + success = True + error_msg = None + + # Save result to parquet file and compute hash + result_hash = None + if result is None: + raise ValueError("Operation returned None, which is not allowed.") + + # Create outputs/results directory if it doesn't exist + results_dir = "outputs/results" + os.makedirs(results_dir, exist_ok=True) + + # Save result to parquet file + output_filename = f"{results_dir}/{operation_name}_{framework}.parquet" + + if hasattr(result, "to_frame"): + result = result.to_frame(name=operation_name) + + if hasattr(result, "to_parquet"): + result.to_parquet(output_filename, index=False) + elif hasattr(result, "write_parquet"): + result.write_parquet(output_filename) + else: + # Handle other types (scalars, arrays, etc.) + # by converting to DataFrame + if hasattr(result, "__iter__") and not isinstance(result, (str, bytes)): + # Iterable but not string + temp_df = df_lib.DataFrame({"result": list(result)}) + else: + # Scalar value + temp_df = df_lib.DataFrame({"result": [result]}) + temp_df.to_parquet(output_filename, index=False) + + # Compute hash of the saved file for consistency verification + with open(output_filename, "rb") as f: + result_hash = hashlib.sha256(f.read()).hexdigest() + + return { + "operation": operation_name, + "framework": framework, + "execution_time": execution_time, + "success": success, + "error": error_msg, + "timestamp": datetime.now(), + "result_hash": result_hash, + } diff --git a/01_prep_data.py b/01_prep_data.py new file mode 100644 index 0000000..9756612 --- /dev/null +++ b/01_prep_data.py @@ -0,0 +1,206 @@ +import os +from typing import Any, Dict + +import numpy as np +import pandas as pd + + +def create_datasets() -> None: + """Create comprehensive datasets for benchmarking pandas, fireducks, and polars""" + + # Create data directory if it doesn't exist + os.makedirs("data", exist_ok=True) + + # Set random seed for reproducibility + np.random.seed(42) + + # Dataset sizes - make them large enough for meaningful benchmarks + n_customers = 100_000 + n_orders = 500_000 + n_products = 10_000 + n_order_items = 1_200_000 + n_reviews = 300_000 + + print("Creating customers dataset...") + # Customers table + customers = pd.DataFrame( + { + "customer_id": range(1, n_customers + 1), + "name": [f"Customer_{i}" for i in range(1, n_customers + 1)], + "email": [f"customer_{i}@email.com" for i in range(1, n_customers + 1)], + "age": np.random.randint(18, 80, n_customers), + "city": np.random.choice( + [ + "New York", + "Los Angeles", + "Chicago", + "Houston", + "Phoenix", + "Philadelphia", + "San Antonio", + "San Diego", + "Dallas", + "Austin", + ], + n_customers, + ), + "registration_date": pd.date_range( + "2020-01-01", periods=n_customers, freq="1h" + ), + "annual_income": np.random.normal(50000, 20000, n_customers), + "customer_segment": np.random.choice( + ["Premium", "Standard", "Basic"], n_customers, p=[0.2, 0.5, 0.3] + ), + } + ) + + print("Creating products dataset...") + # Products table + products = pd.DataFrame( + { + "product_id": range(1, n_products + 1), + "product_name": [f"Product_{i}" for i in range(1, n_products + 1)], + "category": np.random.choice( + ["Electronics", "Clothing", "Books", "Home", "Sports"], n_products + ), + "price": np.random.uniform(10, 1000, n_products), + "cost": np.random.uniform(5, 500, n_products), + "weight": np.random.uniform(0.1, 50, n_products), + "supplier_id": np.random.randint(1, 1000, n_products), + "in_stock": np.random.choice([True, False], n_products, p=[0.8, 0.2]), + } + ) + + print("Creating orders dataset...") + # Orders table + orders = pd.DataFrame( + { + "order_id": range(1, n_orders + 1), + "customer_id": np.random.randint(1, n_customers + 1, n_orders), + "order_date": pd.date_range("2020-01-01", periods=n_orders, freq="30min"), + "shipping_date": pd.date_range( + "2020-01-02", periods=n_orders, freq="30min" + ), + "total_amount": np.random.uniform(20, 5000, n_orders), + "discount_amount": np.random.uniform(0, 500, n_orders), + "shipping_cost": np.random.uniform(5, 50, n_orders), + "status": np.random.choice( + ["Pending", "Shipped", "Delivered", "Cancelled"], + n_orders, + p=[0.1, 0.2, 0.6, 0.1], + ), + "payment_method": np.random.choice( + ["Credit Card", "PayPal", "Bank Transfer"], n_orders + ), + } + ) + + print("Creating order items dataset...") + # Order Items table (junction table) + order_items = pd.DataFrame( + { + "order_item_id": range(1, n_order_items + 1), + "order_id": np.random.randint(1, n_orders + 1, n_order_items), + "product_id": np.random.randint(1, n_products + 1, n_order_items), + "quantity": np.random.randint(1, 10, n_order_items), + "unit_price": np.random.uniform(10, 1000, n_order_items), + "discount_percentage": np.random.uniform(0, 0.3, n_order_items), + } + ) + + print("Creating reviews dataset...") + # Reviews table + reviews = pd.DataFrame( + { + "review_id": range(1, n_reviews + 1), + "customer_id": np.random.randint(1, n_customers + 1, n_reviews), + "product_id": np.random.randint(1, n_products + 1, n_reviews), + "order_id": np.random.randint(1, n_orders + 1, n_reviews), + "rating": np.random.randint(1, 6, n_reviews), + "review_text": [f"Review text {i}" for i in range(1, n_reviews + 1)], + "review_date": pd.date_range("2020-02-01", periods=n_reviews, freq="2h"), + "helpful_votes": np.random.randint(0, 100, n_reviews), + } + ) + + print("Creating time series dataset...") + # Time series data for more complex operations + dates = pd.date_range("2020-01-01", "2023-12-31", freq="D") + time_series = pd.DataFrame( + { + "date": dates, + "sales": np.random.normal(10000, 2000, len(dates)) + + 1000 + * np.sin(2 * np.pi * np.arange(len(dates)) / 365.25), # seasonal pattern + "marketing_spend": np.random.normal(5000, 1000, len(dates)), + "temperature": 20 + + 15 * np.sin(2 * np.pi * np.arange(len(dates)) / 365.25) + + np.random.normal(0, 5, len(dates)), + "website_visits": np.random.poisson(50000, len(dates)), + "conversion_rate": np.random.beta(2, 8, len(dates)), + } + ) + + # Save all datasets as parquet files + print("Saving datasets to parquet files...") + customers.to_parquet("data/customers.parquet", index=False) + products.to_parquet("data/products.parquet", index=False) + orders.to_parquet("data/orders.parquet", index=False) + order_items.to_parquet("data/order_items.parquet", index=False) + reviews.to_parquet("data/reviews.parquet", index=False) + time_series.to_parquet("data/time_series.parquet", index=False) + + # Create some additional datasets with different characteristics + print("Creating additional datasets for diverse benchmarks...") + + # Wide dataset (many columns) + wide_data = pd.DataFrame({f"col_{i}": np.random.randn(50000) for i in range(100)}) + wide_data["id"] = range(50000) + wide_data.to_parquet("data/wide_data.parquet", index=False) + + # Text-heavy dataset + text_data = pd.DataFrame( + { + "id": range(100000), + "text_col_1": [ + f"This is a long text string number {i} with many words" * 10 + for i in range(100000) + ], + "text_col_2": [f"Another text column {i}" * 5 for i in range(100000)], + "category": np.random.choice(["A", "B", "C", "D", "E"], 100000), + "value": np.random.randn(100000), + } + ) + text_data.to_parquet("data/text_data.parquet", index=False) + + print("Data preparation completed successfully!") + print("Created datasets:") + print(f"- customers: {len(customers):,} rows") + print(f"- products: {len(products):,} rows") + print(f"- orders: {len(orders):,} rows") + print(f"- order_items: {len(order_items):,} rows") + print(f"- reviews: {len(reviews):,} rows") + print(f"- time_series: {len(time_series):,} rows") + print(f"- wide_data: {len(wide_data):,} rows") + print(f"- text_data: {len(text_data):,} rows") + + +def load_data(df_lib: Any) -> Dict[str, Any]: + """Load all datasets""" + print(f"Loading data with {df_lib.__name__}") + data_files = [ + "customers", + "products", + "orders", + "order_items", + "reviews", + "time_series", + "wide_data", + "text_data", + ] + data = {name: df_lib.read_parquet(f"data/{name}.parquet") for name in data_files} + return data + + +if __name__ == "__main__": + create_datasets() diff --git a/02_benchmark.py b/02_benchmark.py new file mode 100644 index 0000000..63c7909 --- /dev/null +++ b/02_benchmark.py @@ -0,0 +1,362 @@ +# Standard imports: +import importlib +import os +import sys +from types import ModuleType +from typing import Any, Callable, Dict, List + +# Local imports using importlib for numbered modules +tools_module = importlib.import_module("00_tools") +prep_data_module = importlib.import_module("01_prep_data") +# Import specific functions from the modules +time_operation: Callable[[str, ModuleType, Callable[..., Any]], Dict[str, Any]] = ( + getattr(tools_module, "time_operation") +) +load_data: Callable[[ModuleType], Dict[str, Any]] = getattr( + prep_data_module, "load_data" +) + +# Dynamic import based on command line argument +if "fireducks" in sys.argv: + import fireducks.pandas as df_lib + + framework = "fireducks" + +elif "pandas" in sys.argv: + import pandas as df_lib + + framework = "pandas" + +else: + raise ValueError("Please specify 'pandas' or 'fireducks' as argument") + + +def run_benchmarks() -> List[Dict[str, Any]]: + """Run comprehensive benchmarks""" + results = [] + + data = load_data(df_lib) + + # Extract dataframes + customers = data["customers"] + products = data["products"] + orders = data["orders"] + order_items = data["order_items"] + reviews = data["reviews"] + time_series = data["time_series"] + wide_data = data["wide_data"] + text_data = data["text_data"] + + print(f"Running benchmarks with {framework}...") + + # Basic operations + results.append( + time_operation( + "basic_filtering", df_lib, lambda: customers[customers["age"] > 30] + ) + ) + + def groupby_aggregation_operation(): + # Ensure consistent groupby behavior by sorting result + result = customers.groupby("city")["annual_income"].agg( + ["mean", "std", "count"] + ) + return result.sort_index() + + results.append( + time_operation( + "groupby_aggregation", + df_lib, + groupby_aggregation_operation, + ) + ) + + results.append( + time_operation( + "sorting", + df_lib, + lambda: orders.sort_values( + ["order_date", "total_amount"], ascending=[True, False] + ), + ) + ) + + # Join operations (multiple types) + results.append( + time_operation( + "simple_inner_join", + df_lib, + lambda: orders.merge(customers, on="customer_id", how="inner"), + ) + ) + + results.append( + time_operation( + "left_join", + df_lib, + lambda: orders.merge(customers, on="customer_id", how="left"), + ) + ) + + results.append( + time_operation( + "complex_multi_join", + df_lib, + lambda: orders.merge(customers, on="customer_id") + .merge(order_items, on="order_id") + .merge(products, on="product_id") + .sort_values(["order_id", "order_item_id"]) + .reset_index(drop=True), + ) + ) + + results.append( + time_operation( + "four_table_join", + df_lib, + lambda: customers.merge(orders, on="customer_id") + .merge(order_items, on="order_id") + .merge(products, on="product_id") + .merge(reviews, on=["customer_id", "product_id"]) + .sort_values("customer_id") + .reset_index(drop=True), + ) + ) + + # Window functions + results.append( + time_operation( + "window_functions", + df_lib, + lambda: orders.assign( + running_total=orders.groupby("customer_id")["total_amount"].cumsum(), + rank=orders.groupby("customer_id")["total_amount"].rank(method="dense"), + ), + ) + ) + + # String operations + results.append( + time_operation( + "string_operations", + df_lib, + lambda: text_data.assign( + text_length=text_data["text_col_1"].str.len(), + text_upper=text_data["text_col_1"].str.upper(), + contains_number=text_data["text_col_1"].str.contains(r"\d+"), + ), + ) + ) + + # Datetime operations + results.append( + time_operation( + "datetime_operations", + df_lib, + lambda: orders.assign( + year=orders["order_date"].dt.year, + month=orders["order_date"].dt.month, + day_of_week=orders["order_date"].dt.dayofweek, + days_to_ship=(orders["shipping_date"] - orders["order_date"]).dt.days, + ), + ) + ) + + # Complex aggregations + def complex_groupby_operation(): + # Ensure consistent groupby behavior by explicitly handling missing groups + result = orders.groupby(["status", orders["order_date"].dt.year]).agg( + { + "total_amount": ["sum", "mean", "count"], + "discount_amount": ["sum", "mean"], + "shipping_cost": "mean", + } + ) + # Sort by index to ensure consistent ordering + return result.sort_index() + + results.append( + time_operation( + "complex_groupby", + df_lib, + complex_groupby_operation, + ) + ) + + # Pivot operations + results.append( + time_operation( + "pivot_table", + df_lib, + lambda: orders.pivot_table( + values="total_amount", + index="customer_id", + columns="status", + aggfunc=["sum", "count"], + fill_value=0, + ), + ) + ) + + # Statistical operations + results.append( + time_operation( + "statistical_operations", + df_lib, + lambda: customers.select_dtypes(include=["number"]).describe(), + ) + ) + + def correlation_matrix_operation(): + # Ensure consistent correlation matrix by explicitly dropping NaN and sorting + numeric_data = time_series.select_dtypes(include=["number"]).dropna() + corr_matrix = numeric_data.corr() + # Fill diagonal with 1.0 explicitly to ensure consistency + for i in range(len(corr_matrix)): + corr_matrix.iloc[i, i] = 1.0 + return corr_matrix.sort_index().sort_index(axis=1) + + results.append( + time_operation( + "correlation_matrix", + df_lib, + correlation_matrix_operation, + ) + ) + + # Rolling window operations + def rolling_operations_func(): + # Ensure consistent rolling operations by explicitly handling NaN values + result = time_series.assign( + sales_ma_7=time_series["sales"].rolling(window=7, min_periods=7).mean(), + sales_ma_30=time_series["sales"].rolling(window=30, min_periods=30).mean(), + sales_std_7=time_series["sales"].rolling(window=7, min_periods=7).std(), + ) + return result + + results.append( + time_operation( + "rolling_operations", + df_lib, + rolling_operations_func, + ) + ) + + results.append( + time_operation("wide_data_transpose", df_lib, lambda: wide_data.head(1000).T) + ) + + # Memory intensive operations + results.append( + time_operation( + "large_concat", + df_lib, + lambda: df_lib.concat([customers] * 5, ignore_index=True), + ) + ) + + # Advanced joins with conditions + results.append( + time_operation( + "conditional_join", + df_lib, + lambda: customers.merge(orders, on="customer_id") + .query("age > 25 and total_amount > 100") + .sort_values("customer_id") + .reset_index(drop=True), + ) + ) + + # Complex filtering + results.append( + time_operation( + "complex_filtering", + df_lib, + lambda: orders[ + (orders["total_amount"] > orders["total_amount"].quantile(0.75)) + & (orders["status"] == "Delivered") + & (orders["order_date"] >= "2021-01-01") + ], + ) + ) + + # Cross tabulation (if supported) + crosstab_func = getattr(df_lib, "crosstab", None) + if crosstab_func is not None: + results.append( + time_operation( + "crosstab", + df_lib, + lambda: crosstab_func(customers["city"], customers["customer_segment"]), + ) + ) + + # Multi-level groupby + results.append( + time_operation( + "multilevel_groupby", + df_lib, + lambda: order_items.groupby(["order_id", "product_id"]).agg( + {"quantity": "sum", "unit_price": "mean", "discount_percentage": "max"} + ), + ) + ) + + # Time series resampling + results.append( + time_operation( + "time_series_resample", + df_lib, + lambda: time_series.set_index("date") + .resample("ME") + .agg({"sales": "sum", "marketing_spend": "sum", "website_visits": "mean"}), + ) + ) + + # Quantile operations + results.append( + time_operation( + "quantile_operations", + df_lib, + lambda: customers.groupby("customer_segment")["annual_income"].quantile( + [0.25, 0.5, 0.75] + ), + ) + ) + + return results + + +def main() -> None: + # Create outputs directory + os.makedirs("outputs", exist_ok=True) + + cache_used = "--cache" in sys.argv + + # Run benchmarks + results = run_benchmarks() + + # Convert results to DataFrame + results_df = df_lib.DataFrame(results) + results_df["cache_used"] = cache_used + + # Save results + cache_suffix = "_cache" if cache_used else "_no_cache" + output_file = f"outputs/{framework}{cache_suffix}_results.parquet" + results_df.to_parquet(output_file, index=False) + + print(f"Benchmarks completed. Results saved to {output_file}") + print(f"Total operations: {len(results)}") + print(f"Successful operations: {sum(1 for r in results if r['success'])}") + print(f"Failed operations: {sum(1 for r in results if not r['success'])}") + + # Print summary statistics + successful_results = [r for r in results if r["success"]] + if successful_results: + times = [r["execution_time"] for r in successful_results] + print(f"Average execution time: {sum(times) / len(times):.4f} seconds") + print(f"Total execution time: {sum(times):.4f} seconds") + + +if __name__ == "__main__": + main() diff --git a/03_polars.py b/03_polars.py new file mode 100644 index 0000000..9863d58 --- /dev/null +++ b/03_polars.py @@ -0,0 +1,443 @@ +# Standard imports +import argparse +import importlib.util +import os +from typing import Any, Callable, Dict, List + +import pandas as pd + +# PyPI imports +import polars as pl + +# Local imports using importlib for numbered modules +tools_module = importlib.import_module("00_tools") +prep_data_module = importlib.import_module("01_prep_data") +# Import specific functions from the modules +time_operation: Callable[[str, Any, Callable[..., Any]], Dict[str, Any]] = getattr( + tools_module, "time_operation" +) +load_data: Callable[[Any], Dict[str, Any]] = getattr(prep_data_module, "load_data") + + +def run_benchmarks( + data: Dict[str, Any], use_cache: bool = False +) -> List[Dict[str, Any]]: + """Run comprehensive benchmarks using polars (including lazy evaluation)""" + results = [] + + # Extract dataframes + customers = data["customers"] + products = data["products"] + orders = data["orders"] + order_items = data["order_items"] + reviews = data["reviews"] + time_series = data["time_series"] + wide_data = data["wide_data"] + text_data = data["text_data"] + + print("Running benchmarks with polars...") + + # Basic operations (eager) + results.append( + time_operation( + "basic_filtering", + pl, + lambda: customers.filter(pl.col("age") > 30), + ) + ) + + def groupby_aggregation_polars(): + # Use Polars native group_by and aggregation + result_pl = ( + customers.group_by("city") + .agg( + pl.mean("annual_income").alias("mean"), + pl.std("annual_income").alias("std"), + pl.count("annual_income").alias("count"), + ) + .sort("city") + ) + # Convert to pandas DataFrame with city as index to match pandas output + return result_pl.to_pandas().set_index("city") + + results.append( + time_operation( + "groupby_aggregation", + pl, + groupby_aggregation_polars, + ) + ) + + results.append( + time_operation( + "sorting", + pl, + lambda: orders.sort( + ["order_date", "total_amount"], descending=[False, True] + ), + ) + ) + + # Join operations + results.append( + time_operation( + "simple_inner_join", + pl, + lambda: orders.join(customers, on="customer_id", how="inner"), + ) + ) + + results.append( + time_operation( + "left_join", + pl, + lambda: orders.join(customers, on="customer_id", how="left"), + ) + ) + + def complex_multi_join_polars(): + result = ( + orders.join(customers, on="customer_id") + .join(order_items, on="order_id") + .join(products, on="product_id") + .sort(["order_id", "order_item_id"]) + .to_pandas() + .reset_index(drop=True) + ) + return result + + results.append( + time_operation( + "complex_multi_join", + pl, + complex_multi_join_polars, + ) + ) + + def four_table_join_polars(): + # Perform the joins step by step to control column naming like pandas + result = ( + customers.join(orders, on="customer_id", suffix="_orders") + .join(order_items, on="order_id", suffix="_items") + .join(products, on="product_id", suffix="_products") + .join(reviews, on=["customer_id", "product_id"], suffix="_reviews") + .sort("customer_id") # Add sorting to ensure consistent order + ) + + # Convert to pandas and rename columns to match pandas merge behavior + df = result.to_pandas() + + # Rename columns to match pandas naming convention + if "order_id_reviews" in df.columns: + df = df.rename(columns={"order_id_reviews": "order_id_y"}) + if "order_id" in df.columns: + df = df.rename(columns={"order_id": "order_id_x"}) + + return df + + results.append(time_operation("four_table_join", pl, four_table_join_polars)) + + # Window functions + def window_functions_polars(): + # Use Polars native window functions. + # Cast rank to float to match pandas output dtype. + result = orders.with_columns( + pl.col("total_amount").cum_sum().over("customer_id").alias("running_total"), + pl.col("total_amount") + .rank(method="dense") + .over("customer_id") + .cast(pl.Float64) + .alias("rank"), + ) + return result + + results.append(time_operation("window_functions", pl, window_functions_polars)) + + # String operations + def string_operations_polars(): + result = text_data.with_columns( + [ + pl.col("text_col_1") + .str.len_chars() + .cast(pl.Int64) + .alias("text_length"), # Cast to match pandas int64 + pl.col("text_col_1").str.to_uppercase().alias("text_upper"), + pl.col("text_col_1").str.contains(r"\d+").alias("contains_number"), + ] + ).to_pandas() + return result + + results.append(time_operation("string_operations", pl, string_operations_polars)) + + # Datetime operations + def datetime_operations_polars(): + result = orders.with_columns( + [ + pl.col("order_date").dt.year().alias("year"), + pl.col("order_date") + .dt.month() + .cast(pl.Int32) + .alias("month"), # Cast to match pandas int32 + (pl.col("order_date").dt.weekday() - 1) + .cast(pl.Int32) + .alias("day_of_week"), # Convert Monday=1 to Monday=0 to match pandas + (pl.col("shipping_date") - pl.col("order_date")) + .dt.total_days() + .alias("days_to_ship"), + ] + ).to_pandas() + return result + + results.append( + time_operation("datetime_operations", pl, datetime_operations_polars) + ) + + # Complex aggregations + def complex_groupby_polars(): + # Use Polars native group_by operations + result_pl = ( + orders.with_columns(pl.col("order_date").dt.year().alias("year")) + .group_by(["status", "year"]) + .agg( + [ + pl.col("total_amount").sum().alias("total_amount_sum"), + pl.col("total_amount").mean().alias("total_amount_mean"), + pl.col("total_amount").count().alias("total_amount_count"), + pl.col("discount_amount").sum().alias("discount_amount_sum"), + pl.col("discount_amount").mean().alias("discount_amount_mean"), + pl.col("shipping_cost").mean().alias("shipping_cost_mean"), + ] + ) + .sort(["status", "year"]) + ) + + # Convert to pandas and reshape to match pandas groupby format with MultiIndex + df = result_pl.to_pandas() + + # Create the MultiIndex structure that pandas groupby produces + + # Reshape data to match pandas multi-level column format + data = {} + data[("total_amount", "sum")] = df["total_amount_sum"] + data[("total_amount", "mean")] = df["total_amount_mean"] + data[("total_amount", "count")] = df["total_amount_count"] + data[("discount_amount", "sum")] = df["discount_amount_sum"] + data[("discount_amount", "mean")] = df["discount_amount_mean"] + data[("shipping_cost", "mean")] = df["shipping_cost_mean"] + + # Create result DataFrame with MultiIndex columns and MultiIndex index + result_df = pd.DataFrame(data) + result_df.index = pd.MultiIndex.from_arrays( + [df["status"], df["year"]], names=["status", "year"] + ) + + return result_df.sort_index() + + results.append(time_operation("complex_groupby", pl, complex_groupby_polars)) + + # Pivot operations (using polars pivot) + def pivot_table_polars(): + # Use pandas logic for consistent results + pandas_orders = orders.to_pandas() + result = pandas_orders.pivot_table( + values="total_amount", + index="customer_id", + columns="status", + aggfunc=["sum", "count"], + fill_value=0, + ) + return result + + results.append(time_operation("pivot_table", pl, pivot_table_polars)) + + # Statistical operations + def statistical_operations_polars(): + # Use pandas logic for consistent results + pandas_customers = customers.to_pandas() + result = pandas_customers.select_dtypes(include=["number"]).describe() + return result + + results.append( + time_operation("statistical_operations", pl, statistical_operations_polars) + ) + + # Correlation matrix (select numeric columns) + def correlation_matrix_polars(): + # Use pandas logic for consistent results + pandas_time_series = time_series.to_pandas() + numeric_data = pandas_time_series.select_dtypes(include=["number"]).dropna() + corr_matrix = numeric_data.corr() + # Fill diagonal with 1.0 explicitly to ensure consistency + for i in range(len(corr_matrix)): + corr_matrix.iloc[i, i] = 1.0 + return corr_matrix.sort_index().sort_index(axis=1) + + results.append( + time_operation( + "correlation_matrix", + pl, + correlation_matrix_polars, + ) + ) + + # Rolling window operations + def rolling_operations_polars(): + # Use pandas logic for consistent results and to avoid deprecation warnings + pandas_time_series = time_series.to_pandas() + result = pandas_time_series.assign( + sales_ma_7=pandas_time_series["sales"] + .rolling(window=7, min_periods=7) + .mean(), + sales_ma_30=pandas_time_series["sales"] + .rolling(window=30, min_periods=30) + .mean(), + sales_std_7=pandas_time_series["sales"] + .rolling(window=7, min_periods=7) + .std(), + ) + return result + + results.append(time_operation("rolling_operations", pl, rolling_operations_polars)) + + def wide_data_transpose_polars(): + # Use pandas logic for consistent results + pandas_wide_data = wide_data.to_pandas() + result = pandas_wide_data.head(1000).T + return result + + results.append( + time_operation( + "wide_data_transpose", + pl, + wide_data_transpose_polars, + ) + ) + + # Memory intensive operations + results.append( + time_operation("large_concat", pl, lambda: pl.concat([customers] * 5)) + ) + + # Advanced filtering + def conditional_join_polars(): + result = ( + customers.join(orders, on="customer_id") + .filter((pl.col("age") > 25) & (pl.col("total_amount") > 100)) + .sort("customer_id") + .to_pandas() + .reset_index(drop=True) + ) + return result + + results.append( + time_operation( + "conditional_join", + pl, + conditional_join_polars, + ) + ) + + # Complex filtering + results.append( + time_operation( + "complex_filtering", + pl, + lambda: orders.filter( + (pl.col("total_amount") > orders["total_amount"].quantile(0.75)) + & (pl.col("status") == "Delivered") + & (pl.col("order_date") >= pl.datetime(2021, 1, 1)) + ).to_pandas(), + ) + ) + + # Cross tabulation (using group_by and pivot) + def crosstab_polars(): + # Use pandas crosstab for consistent results + pandas_customers = customers.to_pandas() + result = pd.crosstab( + pandas_customers["city"], pandas_customers["customer_segment"] + ) + return result + + results.append(time_operation("crosstab", pl, crosstab_polars)) + + # Multi-level groupby + def multilevel_groupby_polars(): + # Use pandas logic for consistent results + pandas_order_items = order_items.to_pandas() + result = pandas_order_items.groupby(["order_id", "product_id"]).agg( + {"quantity": "sum", "unit_price": "mean", "discount_percentage": "max"} + ) + return result + + results.append(time_operation("multilevel_groupby", pl, multilevel_groupby_polars)) + + # Time series resampling + def time_series_resample_polars(): + # Use pandas logic for consistent results + pandas_time_series = time_series.to_pandas() + result = ( + pandas_time_series.set_index("date") + .resample("ME") + .agg({"sales": "sum", "marketing_spend": "sum", "website_visits": "mean"}) + ) + return result + + results.append( + time_operation("time_series_resample", pl, time_series_resample_polars) + ) + + # Quantile operations + def quantile_operations_polars(): + # Use pandas logic for consistent results + pandas_customers = customers.to_pandas() + result = pandas_customers.groupby("customer_segment")["annual_income"].quantile( + [0.25, 0.5, 0.75] + ) + return result + + results.append( + time_operation("quantile_operations", pl, quantile_operations_polars) + ) + + return results + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--cache", action="store_true", help="Use cached data") + args = parser.parse_args() + + # Create outputs directory + os.makedirs("outputs", exist_ok=True) + + # Load data + data = load_data(pl) + + # Run benchmarks + results = run_benchmarks(data, use_cache=args.cache) + + # Convert results to pandas DataFrame for consistency + results_df = pl.DataFrame(results).to_pandas() + results_df["cache_used"] = args.cache + + # Save results + cache_suffix = "_cache" if args.cache else "_no_cache" + output_file = f"outputs/polars{cache_suffix}_results.parquet" + results_df.to_parquet(output_file, index=False) + + print(f"Benchmarks completed. Results saved to {output_file}") + print(f"Total operations: {len(results)}") + print(f"Successful operations: {sum(1 for r in results if r['success'])}") + print(f"Failed operations: {sum(1 for r in results if not r['success'])}") + + # Print summary statistics + successful_results = [r for r in results if r["success"]] + if successful_results: + times = [r["execution_time"] for r in successful_results] + print(f"Average execution time: {sum(times) / len(times):.4f} seconds") + print(f"Total execution time: {sum(times):.4f} seconds") + + +if __name__ == "__main__": + main() diff --git a/04_compare_parquets.py b/04_compare_parquets.py new file mode 100644 index 0000000..b4ee673 --- /dev/null +++ b/04_compare_parquets.py @@ -0,0 +1,140 @@ +from pathlib import Path + +import pandas as pd +import pandas.testing as pd_testing + +# Define the relative path to the results folder +folder_path = Path(__file__).parent / "outputs" / "results" + + +def group_files_by_operation(path): + """Groups parquet files by their base operation name.""" + # Define supported frameworks + frameworks = ["fireducks", "pandas", "polars"] + + files_grouped = {} + for file in path.glob("*.parquet"): + stem = file.stem + + # Find which framework this file belongs to + framework_found = None + operation_key = None + + for framework in frameworks: + suffix = f"_{framework}" + if stem.endswith(suffix): + framework_found = framework + operation_key = stem.removesuffix(suffix) + break + + if framework_found is None: + raise ValueError(f"Unknown or missing framework in file: {file.name}") + + files_grouped.setdefault(operation_key, {})[framework_found] = file + + return files_grouped + + +def compare_dataframes(df1, df2, label1, label2): + """ + Compares two DataFrames using pandas.testing.assert_frame_equal. + + Returns: + bool: True if DataFrames are equal, False otherwise. + """ + print(f"Comparing {label1} vs {label2}:") + + try: + pd_testing.assert_frame_equal( + df1, + df2, + check_dtype=False, # Allow dtype differences like int64 vs uint32 + check_index_type=True, + check_column_type=True, + check_frame_type=True, + check_names=True, + rtol=1e-10, # Relative tolerance for numerical comparisons + atol=1e-12, # Absolute tolerance for numerical comparisons + check_exact=False, # Allow numerical tolerance + ) + print(f"✅ DataFrames are equal between {label1} and {label2}") + print() + return True + + except AssertionError as e: + print(f"⚠️ DataFrames differ between {label1} and {label2}:") + print(f" {str(e)}") + + # Show detailed comparison for debugging with comprehensive checks + try: + # Check that columns match + if (set1 := set(df1.columns)) != (set2 := set(df2.columns)): + print("⚠️ Columns differ:") + print(f" Only in {label1}: {set1 - set2}") + print(f" Only in {label2}: {set2 - set1}") + # Check that length matches + elif len(df1) != len(df2): + print(f"⚠️ Row count differs: {len(df1)} vs {len(df2)}") + else: + # Check that data types match for common columns + type_mismatches = [] + for col in df1.columns: + if df1[col].dtype != df2[col].dtype: + type_mismatches.append((col, df1[col].dtype, df2[col].dtype)) + + if type_mismatches: + print("⚠️ Data type mismatches:") + for col, dtype1, dtype2 in type_mismatches: + print(f" {col}: {dtype1} vs {dtype2}") + else: + # Only try detailed comparison if basic structure matches + diff = df1.compare(df2, align_axis=1) + if not diff.empty: + print("\nDetailed comparison (first 5 rows):") + print(diff.head().to_markdown()) + except Exception as detailed_e: + print(f"\nWarning: Could not generate detailed comparison: {detailed_e}") + + print() + return False + + except Exception as e: + print(f"⚠️ Error during comparison: {e}") + print() + return False + + +# --- Main Execution --- +if not folder_path.exists(): + raise Exception(f"Error: Directory not found at '{folder_path}'") + +# Group files by the test operation +grouped_files = group_files_by_operation(folder_path) + +# Iterate through each operation and perform comparisons +success = True +for key, fdict in sorted(grouped_files.items()): + print(f"--- Comparing results for: {key} ---") + + # Ensure the base pandas file exists for comparison + assert "pandas" in fdict + + df_pandas = pd.read_parquet(fdict["pandas"]) + + # Compare pandas with fireducks + df_fireducks = pd.read_parquet(fdict["fireducks"]) + success &= compare_dataframes( + df_pandas, df_fireducks, f"{key} (pandas)", f"{key} (fireducks)" + ) + + # Compare pandas with polars + # Load polars-generated parquet into a pandas DataFrame for comparison + df_polars = pd.read_parquet(fdict["polars"]) + success &= compare_dataframes( + df_pandas, df_polars, f"{key} (pandas)", f"{key} (polars)" + ) + +if not success: + raise AssertionError( + "One or more benchmark result comparisons failed. Check logs above for details." + ) diff --git a/05_comparison.py b/05_comparison.py new file mode 100644 index 0000000..7ad7e3d --- /dev/null +++ b/05_comparison.py @@ -0,0 +1,496 @@ +from pathlib import Path +from typing import Any, Dict, Optional + +import pandas as pd + + +def load_benchmark_results() -> Dict[str, pd.DataFrame]: + """Load all benchmark result files""" + results_dir = Path("outputs") + + # Expected result files + expected_files = [ + "pandas_no_cache_results.parquet", + "pandas_cache_results.parquet", + "fireducks_no_cache_results.parquet", + "fireducks_cache_results.parquet", + "polars_no_cache_results.parquet", + "polars_cache_results.parquet", + ] + + results = {} + missing_files = [] + + for file in expected_files: + file_path = results_dir / file + if file_path.exists(): + try: + df = pd.read_parquet(file_path) + results[file.replace("_results.parquet", "")] = df + print(f"✓ Loaded {file}") + except Exception as e: + print(f"✗ Error loading {file}: {e}") + missing_files.append(file) + else: + print(f"✗ Missing file: {file}") + missing_files.append(file) + + if missing_files: + print( + f"\nWarning: {len(missing_files)} files are missing. " + f"Analysis will continue with available data." + ) + + return results + + +def analyze_result_hashes(successful_df: pd.DataFrame) -> Optional[Dict[str, Any]]: + """Analyze and compare result hashes across frameworks""" + + # Filter out rows without hashes (failed operations or old runs without + # hash support) + hash_df = successful_df[successful_df["result_hash"].notna()].copy() + + if len(hash_df) == 0: + print( + "No hash data available for comparison. Results may be from " + "older benchmark runs." + ) + return None + + # Create a pivot table of hashes by operation and framework + hash_pivot = hash_df.pivot_table( + index="operation", + columns="framework", + values="result_hash", + aggfunc="first", # Take first hash (should be same for cache/no-cache) + ) + + # Analyze consistency + hash_analysis = [] + + for operation in hash_pivot.index: + operation_hashes = hash_pivot.loc[operation].dropna() + + if len(operation_hashes) > 1: + # Check if all hashes are the same + unique_hashes = operation_hashes.unique() + is_consistent = len(unique_hashes) == 1 + + hash_analysis.append( + { + "operation": operation, + "consistent": is_consistent, + "num_frameworks": len(operation_hashes), + "unique_hashes": len(unique_hashes), + "frameworks": list(operation_hashes.index), + "hashes": dict(operation_hashes), + } + ) + + return { + "hash_pivot": hash_pivot, + "analysis": hash_analysis, + "consistency_summary": { + "total_operations": len(hash_analysis), + "consistent_operations": sum(1 for a in hash_analysis if a["consistent"]), + "inconsistent_operations": sum( + 1 for a in hash_analysis if not a["consistent"] + ), + }, + } + + +def create_comparison_tables( + results: Dict[str, pd.DataFrame], +) -> Optional[Dict[str, Any]]: + """Create comprehensive comparison tables""" + + # Combine all results + all_results = [] + for key, df in results.items(): + df_copy = df.copy() + + # Parse framework and cache info from filename + parts = key.split("_") + framework = parts[0] + cache_status = "cache" if "cache" in parts else "no_cache" + + df_copy["framework"] = framework + df_copy["cache_status"] = cache_status + all_results.append(df_copy) + + if not all_results: + print("No results to analyze!") + return None + + combined_df = pd.concat(all_results, ignore_index=True) + + # Filter only successful operations + successful_df = combined_df[combined_df["success"]].copy() + + if len(successful_df) == 0: + print("No successful operations found!") + return None + + print( + f"\nAnalyzing {len(successful_df)} successful operations across " + f"{successful_df['framework'].nunique()} frameworks" + ) + + # Create pivot tables for different views + + # 1. Main comparison table: Operation vs Framework+Cache + pivot_df = successful_df.pivot_table( + index="operation", + columns=["framework", "cache_status"], + values="execution_time", + aggfunc="mean", + ) + + # 2. Summary statistics by framework + summary_stats = ( + successful_df.groupby(["framework", "cache_status"])["execution_time"] + .agg(["count", "mean", "median", "std", "min", "max", "sum"]) + .round(4) + ) + + # 3. Speed comparison (relative to pandas no cache) + if ("pandas", "no_cache") in pivot_df.columns: + baseline = pivot_df[("pandas", "no_cache")] + speedup_df = pivot_df.div(baseline, axis=0) + speedup_df.columns = [ + f"{fw}_{cache}_speedup" for fw, cache in speedup_df.columns + ] + else: + speedup_df = None + + # 4. Cache effectiveness (speedup from caching) + cache_effectiveness = {} + frameworks = successful_df["framework"].unique() + + for fw in frameworks: + no_cache_col = (fw, "no_cache") + cache_col = (fw, "cache") + + if no_cache_col in pivot_df.columns and cache_col in pivot_df.columns: + effectiveness = pivot_df[no_cache_col] / pivot_df[cache_col] + cache_effectiveness[f"{fw}_cache_speedup"] = effectiveness + + if cache_effectiveness: + cache_effectiveness_df = pd.DataFrame(cache_effectiveness) + else: + cache_effectiveness_df = None + + # 5. Hash comparison for result verification + hash_comparison = analyze_result_hashes(successful_df) + + return { + "main_comparison": pivot_df, + "summary_stats": summary_stats, + "speedup_comparison": speedup_df, + "cache_effectiveness": cache_effectiveness_df, + "raw_data": successful_df, + "hash_comparison": hash_comparison, + } + + +def print_analysis_report(tables: Dict[str, Any]) -> None: + """Print comprehensive analysis report""" + + print("\n" + "=" * 80) + print("DATAFRAME LIBRARY BENCHMARK ANALYSIS REPORT") + print("=" * 80) + + main_comparison = tables["main_comparison"] + summary_stats = tables["summary_stats"] + speedup_comparison = tables["speedup_comparison"] + cache_effectiveness = tables["cache_effectiveness"] + raw_data = tables["raw_data"] + hash_comparison = tables["hash_comparison"] + + # 1. Executive Summary + print("\n1. EXECUTIVE SUMMARY") + print("-" * 40) + + frameworks = raw_data["framework"].unique() + total_operations = len(main_comparison) + + print(f"• Frameworks tested: {', '.join(frameworks)}") + print(f"• Total operations benchmarked: {total_operations}") + print("• Cache configurations: with and without cache") + + # Overall performance ranking + overall_perf = summary_stats.groupby("framework")["mean"].mean().sort_values() + print("\n• Overall Performance Ranking (avg execution time):") + for i, (fw, time) in enumerate(overall_perf.items(), 1): + print(f" {i}. {fw}: {time:.4f} seconds") + + # 2. Detailed Performance Comparison + print("\n\n2. DETAILED PERFORMANCE COMPARISON") + print("-" * 40) + print("\nExecution times (seconds) by operation and framework:") + print(main_comparison.round(4).to_string()) + + # 3. Summary Statistics + print("\n\n3. SUMMARY STATISTICS") + print("-" * 40) + print(summary_stats.to_string()) + + # 4. Result Consistency Analysis (Hash Comparison) + print("\n\n4. RESULT CONSISTENCY ANALYSIS") + print("-" * 40) + + if hash_comparison is not None: + consistency_summary = hash_comparison["consistency_summary"] + print(f"Total operations compared: {consistency_summary['total_operations']}") + print( + f"Consistent results across frameworks: " + f"{consistency_summary['consistent_operations']}" + ) + print( + f"Inconsistent results across frameworks: " + f"{consistency_summary['inconsistent_operations']}" + ) + + if consistency_summary["inconsistent_operations"] > 0: + print("\nINCONSISTENT OPERATIONS (frameworks produce different results):") + print("-" * 60) + + for analysis in hash_comparison["analysis"]: + if not analysis["consistent"]: + print(f"\nOperation: {analysis['operation']}") + print(f" Frameworks tested: {', '.join(analysis['frameworks'])}") + print(f" Unique result hashes: {analysis['unique_hashes']}") + print(" Hash values by framework:") + for fw, hash_val in analysis["hashes"].items(): + print(f" {fw}: {hash_val[:12]}...") + else: + print("\n✓ All operations produce consistent results across frameworks!") + + # Display hash comparison table + print("\n\nHASH COMPARISON TABLE:") + print(hash_comparison["hash_pivot"].to_string()) + + else: + print("No hash data available for comparison.") + print("This may be because:") + print(" - Results are from older benchmark runs without hash support") + print(" - All operations failed") + print(" - Result saving failed during benchmark execution") + + # 5. Speed Comparison (if available) + if speedup_comparison is not None: + print("\n\n5. RELATIVE PERFORMANCE (vs pandas no cache)") + print("-" * 40) + print("Values < 1.0 indicate faster performance than pandas baseline") + print(speedup_comparison.round(3).to_string()) + + # Best and worst performers + print("\n• Best performing operations by framework:") + for col in speedup_comparison.columns: + if col.endswith("_speedup"): + best_ops = speedup_comparison[col].nsmallest(3) + fw_name = col.replace("_speedup", "") + print(f"\n {fw_name}:") + for op, speedup in best_ops.items(): + if not pd.isna(speedup): + print(f" {op}: {speedup:.3f}x") + + # 6. Cache Effectiveness + if cache_effectiveness is not None: + print("\n\n6. CACHE EFFECTIVENESS") + print("-" * 40) + print("Cache speedup ratios (no_cache_time / cache_time):") + print(cache_effectiveness.round(3).to_string()) + + # Average cache effectiveness by framework + print("\n• Average cache effectiveness:") + for col in cache_effectiveness.columns: + fw_name = col.replace("_cache_speedup", "") + avg_speedup = cache_effectiveness[col].mean() + if not pd.isna(avg_speedup): + print(f" {fw_name}: {avg_speedup:.3f}x speedup from caching") + + # 7. Operation Analysis + print("\n\n7. OPERATION-SPECIFIC ANALYSIS") + print("-" * 40) + + # Most time-consuming operations + op_times = ( + raw_data.groupby("operation")["execution_time"] + .mean() + .sort_values(ascending=False) + ) + print("\n• Most time-consuming operations:") + for i, (op, time) in enumerate(op_times.head(5).items(), 1): + print(f" {i}. {op}: {time:.4f} seconds") + + # Fastest operations + print("\n• Fastest operations:") + for i, (op, time) in enumerate(op_times.tail(5).items(), 1): + print(f" {i}. {op}: {time:.4f} seconds") + + # 8. Framework-Specific Insights + print("\n\n8. FRAMEWORK-SPECIFIC INSIGHTS") + print("-" * 40) + + for fw in frameworks: + fw_data = raw_data[raw_data["framework"] == fw] + failed_ops = fw_data[~fw_data["success"]]["operation"].unique() + + print(f"\n• {fw.upper()}:") + print(f" - Successful operations: {len(fw_data[fw_data['success']])}") + if len(failed_ops) > 0: + print(f" - Failed operations: {len(failed_ops)} ({', '.join(failed_ops)})") + else: + print(" - Failed operations: 0") + + # Performance characteristics + fw_success = fw_data[fw_data["success"]] + if len(fw_success) > 0: + _mean_t = fw_success["execution_time"].mean() + print(f" - Average execution time: {_mean_t:.4f}s") + print( + f" - Median execution time: " + f"{fw_success['execution_time'].median():.4f}s" + ) + print( + f" - Performance consistency (std/mean): " + f"{fw_success['execution_time'].std() / _mean_t:.3f}" + ) + + # 9. Recommendations + print("\n\n9. RECOMMENDATIONS") + print("-" * 40) + + if len(frameworks) >= 2: + fastest_fw = overall_perf.index[0] + print( + f"• For overall performance: {fastest_fw} shows the best " + f"average performance" + ) + + if cache_effectiveness is not None: + best_cache_fw = None + best_cache_speedup = 0 + for col in cache_effectiveness.columns: + fw_name = col.replace("_cache_speedup", "") + avg_speedup = cache_effectiveness[col].mean() + if not pd.isna(avg_speedup) and avg_speedup > best_cache_speedup: + best_cache_speedup = avg_speedup + best_cache_fw = fw_name + + if best_cache_fw: + print( + f"• For cache effectiveness: {best_cache_fw} benefits most " + f"from caching ({best_cache_speedup:.2f}x speedup)" + ) + + print("• Consider workload characteristics when choosing a framework:") + print(" - For complex joins: Check join operation performance") + print(" - For time series: Check rolling and resampling operations") + print(" - For large datasets: Consider memory efficiency and lazy evaluation") + + +def generate_summary_statistics_markdown(summary_stats: pd.DataFrame) -> str: + """Generate markdown formatted summary statistics""" + + markdown = "# Summary Statistics\n\n" + markdown += "## Framework Performance Overview\n\n" + + # Convert to markdown table + markdown += summary_stats.to_markdown() + "\n\n" + + # Add some interpretation + markdown += "## Key Metrics Explanation\n\n" + markdown += "- **count**: Number of operations benchmarked\n" + markdown += "- **mean**: Average execution time (seconds)\n" + markdown += "- **median**: Median execution time (seconds)\n" + markdown += "- **std**: Standard deviation of execution times\n" + markdown += "- **min**: Fastest operation time (seconds)\n" + markdown += "- **max**: Slowest operation time (seconds)\n" + markdown += "- **sum**: Total execution time for all operations (seconds)\n\n" + + # Add ranking based on mean performance + mean_times = summary_stats.groupby("framework")["mean"].mean().sort_values() + markdown += "## Performance Ranking (by average execution time)\n\n" + for i, (framework, avg_time) in enumerate(mean_times.items(), 1): + markdown += f"{i}. **{framework}**: {avg_time:.4f} seconds\n" + + return markdown + + +def save_results_to_files(tables: Dict[str, Any]) -> None: + """Save analysis results to files""" + + output_dir = Path("outputs") + + # Save main comparison table + tables["main_comparison"].to_csv(output_dir / "comparison_table.csv") + print("\n✓ Saved comparison table to outputs/comparison_table.csv") + + # Save summary statistics + tables["summary_stats"].to_csv(output_dir / "summary_statistics.csv") + print("✓ Saved summary statistics to outputs/summary_statistics.csv") + + # Save summary statistics as markdown + summary_markdown = generate_summary_statistics_markdown(tables["summary_stats"]) + with open(output_dir / "summary_statistics.md", "w") as f: + f.write(summary_markdown) + print("✓ Saved summary statistics to outputs/summary_statistics.md") + + # Save speedup comparison if available + if tables["speedup_comparison"] is not None: + tables["speedup_comparison"].to_csv(output_dir / "speedup_comparison.csv") + print("✓ Saved speedup comparison to outputs/speedup_comparison.csv") + + # Save cache effectiveness if available + if tables["cache_effectiveness"] is not None: + tables["cache_effectiveness"].to_csv(output_dir / "cache_effectiveness.csv") + print("✓ Saved cache effectiveness to outputs/cache_effectiveness.csv") + + # Save raw processed data + tables["raw_data"].to_csv(output_dir / "processed_raw_data.csv", index=False) + print("✓ Saved processed raw data to outputs/processed_raw_data.csv") + + # Save hash comparison if available + if tables["hash_comparison"] is not None: + tables["hash_comparison"]["hash_pivot"].to_csv( + output_dir / "hash_comparison.csv" + ) + print("✓ Saved hash comparison to outputs/hash_comparison.csv") + + +def main() -> None: + """Main analysis function""" + + print("Starting benchmark analysis...") + + # Load results + results = load_benchmark_results() + + if not results: + print("No benchmark results found. Please run the benchmarks first.") + return + + # Create comparison tables + tables = create_comparison_tables(results) + + if tables is None: + return + + # Print analysis report + print_analysis_report(tables) + + # Save results to files + save_results_to_files(tables) + + print(f"\n{'=' * 80}") + print("ANALYSIS COMPLETE") + print("=" * 80) + print("Check the outputs/ directory for detailed CSV files with all results.") + + +if __name__ == "__main__": + main() diff --git a/README.md b/README.md index 3af395d..7f92c6c 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,25 @@ # dataframe-benchmarking -Benchmarking a few Dataframe python frameworks +Benchmarking a few dataframe Python frameworks. + +This repo was mostly inspired by the claim that [FireDucks](https://fireducks-dev.github.io/) would be a faster drop-in replacement of [Pandas](https://pandas.pydata.org/), I've made some not too complex test cases based on that claim. For more complex or long-running tasks, I usually resort to [Polars](https://pola.rs/), which requires meaningfully different syntax and usage, but is known for being generally faster. + +The goal was to make some tests that would allow a comparison between those frameworks that is realistic while employing a not-so-heavy use case in terms of numbers of manipulations (something that could favor lazy-evaluated frameworks) but using a realistic data amount. I think a developer should always build something like this in house before proposing a tool migration. You are looking for the best tool for your problem, not the best tool for a problem you've never faced yourself. + +This readme does not mention which tool performed the best, as the repo is designed to run on the latest version of the libraries on the latest Python version each time, thus conclusions are expected to change over time. + +# Usage + +With [uv](https://docs.astral.sh/uv/) installed, run the command `bash run.sh`. + +There is a GitHub Action that executes it and provides artifacts for analysis. If deailed analysis is warranted, or to alleviate comparison burden effect, notice that [pyinstrument](https://pyinstrument.readthedocs.io/en/latest/) reports are available on the artifacts. + +# Other notes: +I suspected `__pycache__` or some Just-in-Time (JIT) compilation artifact could have an impact on performance. Removing the `__pycache__` folder could help with the former, but other than repeating operations, I don't see a proper way of testing the latter. + +Exact same results (sometimes not even allowing for numerical discrepancies) is something that in the past was important for me. Hence I wanted to test what is needed to migrate while not having changes in behavior, this introduced a few quirks in the code, but showcases that for now, a few different behaviors can surface between FireDucks and Pandas, those are minor from what I noted. + +Due to the comparison with polars not being so straightforward, conversion to pandas is sometimes used to avoid leaving gaps in the comparisons. While this makes the comparison unfair for polars on a naive interpretation, the current (unmentioned results) makes this fact relatively indifferent for any practical purposes. + +# Contributing + +Pull requests are welcome, so are issues. Notice that most scripts are deliberately made to crash in cases of errors, so that these are solved instead of reported as errors while tainting the statistics comparison. \ No newline at end of file diff --git a/requirements.in b/requirements.in new file mode 100644 index 0000000..b6a17a9 --- /dev/null +++ b/requirements.in @@ -0,0 +1,14 @@ +# Framewokrs: +pandas +polars +fireducks + +# Linting/formatting: +ruff +ty + +# Time instrumentation +pyinstrument + +# Subdependencies: +tabulate \ No newline at end of file diff --git a/run.sh b/run.sh new file mode 100755 index 0000000..bc1d520 --- /dev/null +++ b/run.sh @@ -0,0 +1,148 @@ + +#!/bin/bash +set -euo pipefail + +# Initialize timing variables +declare -A execution_times +declare -a execution_order + +# Function to run a command and time it +run_timed() { + local name="$1" + shift + local cmd="$*" + + echo "Running: $name" + local start_time=$(date +%s.%N) + eval "$cmd" + local exit_code=$? + local end_time=$(date +%s.%N) + + local duration=$(echo "$end_time - $start_time" | bc -l) + execution_times["$name"]=$(printf "%.2f" "$duration") + execution_order+=("$name") + + if [ $exit_code -ne 0 ]; then + echo "Error: $name failed (exit code: $exit_code)" + echo "Benchmark suite cannot continue with incomplete data. Exiting." + exit 1 + fi + + return $exit_code +} + +# Function to run a Python command with pyinstrument profiling +run_python_profiled() { + local name="$1" + local python_script="$2" + local args="${3:-}" + + # Create pyinstrument output directory + mkdir -p outputs/pyinstrument + + # Create a safe filename for the profiling output + local safe_name=$(echo "$name" | sed 's/[^a-zA-Z0-9._-]/_/g') + local profile_output="outputs/pyinstrument/${safe_name}_profile.html" + + # Run with pyinstrument profiling + local full_cmd="pyinstrument -r html -o \"$profile_output\" $python_script $args" + run_timed "$name" "$full_cmd" +} + +# Clean up previous results: +rm -rf artifacts/* data/* outputs/* __pycache__ +mkdir -p artifacts data outputs + +# Set up environment +echo "Setting up virtual environment..." +uv venv +source .venv/bin/activate +uv pip compile requirements.in | uv pip sync - + +# Log environment details +{ + echo "# Python $(python --version 2>&1)" + uv pip freeze +} > artifacts/requirements_$(date +%Y%m%d_%H%M%S).txt + +# Ensure code quality +echo "Running code quality checks..." +ruff format . +ruff check . --select E,F,I --fix +ty check . + + +# Check if input data exists, else create it +if [ ! -d "data" ] || [ -z "$(ls -A data 2>/dev/null)" ]; then + echo "Data directory is empty or doesn't exist. Creating datasets..." + run_python_profiled "Data Preparation" "01_prep_data.py" "" +else + echo "Data directory exists and contains files. Skipping data preparation." +fi + +# Create outputs directory if it doesn't exist +mkdir -p outputs + +# Clear any existing output files to avoid confusion +echo "Clearing previous benchmark results..." +rm -f outputs/*_results.parquet +rm -rf outputs/results + +echo "" +echo "Starting benchmark runs..." +echo "==========================" + +# Run all cases with and without cache +rm -rf __pycache__ +run_python_profiled "Pandas (no cache)" "02_benchmark.py" "pandas" +run_python_profiled "Pandas (with cache)" "02_benchmark.py" "pandas --cache" +rm -rf __pycache__ +run_python_profiled "Fireducks (no cache)" "02_benchmark.py" "fireducks" +run_python_profiled "Fireducks (with cache)" "02_benchmark.py" "fireducks --cache" +rm -rf __pycache__ +run_python_profiled "Polars (no cache)" "03_polars.py" "" +run_python_profiled "Polars (with cache)" "03_polars.py" "--cache" + +echo "" +echo "All benchmarks completed. Running result comparison..." +echo "=====================================================" + +# Compare parquet results for accuracy verification +run_python_profiled "Result Comparison" "04_compare_parquets.py" "" + +echo "" +echo "Running performance analysis..." +echo "===============================" + +# Run analyzer +run_python_profiled "Analysis" "05_comparison.py" "" + +echo "" +echo "===============================================" +echo "BENCHMARK SUITE EXECUTION TIME SUMMARY" +echo "===============================================" + +# Calculate total time +total_time=0 + +# Display execution times in a formatted table +printf "%-25s %10s\n" "Component" "Time (s)" +printf "%-25s %10s\n" "-------------------------" "----------" + +# Iterate through components in execution order +for key in "${execution_order[@]}"; do + printf "%-25s %10s\n" "$key" "${execution_times[$key]}" + total_time=$(echo "$total_time + ${execution_times[$key]}" | bc -l) +done + +printf "%-25s %10s\n" "-------------------------" "----------" +printf "%-25s %10.2f\n" "TOTAL TIME" "$total_time" + +echo "" +echo "Benchmark suite completed successfully!" +echo "Check the outputs/ directory for detailed results." +echo "Key files:" +echo " - comparison_table.csv: Main performance comparison" +echo " - summary_statistics.csv: Statistical summary" +echo " - speedup_comparison.csv: Relative performance metrics" +echo " - cache_effectiveness.csv: Cache performance analysis" \ No newline at end of file