Real Datasets Pandas vs Power BI: A Head-to-Head

import pandas as pd
import numpy as np
import time
from typing import Optional, 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 process_nyc_taxi_data(
    file_path: str = "nyc_taxi_2023.csv",
    sample_frac: Optional[float] = None
) -> pd.DataFrame:
    """
    Load, clean, and aggregate 2023 NYC Taxi dataset using Pandas 2.2.1.

    Args:
        file_path: Path to 1.2GB CSV file (10M rows)
        sample_frac: Optional fraction to sample (for testing)

    Returns:
        Aggregated DataFrame with hourly fare metrics
    """
    start_time = time.time()
    try:
        # Step 1: Load CSV with optimized dtypes to reduce memory usage
        # Specify dtypes explicitly to avoid inference overhead
        dtype_map = {
            "VendorID": "int8",
            "tpep_pickup_datetime": "str",
            "tpep_dropoff_datetime": "str",
            "passenger_count": "int8",
            "trip_distance": "float32",
            "RatecodeID": "int8",
            "store_and_fwd_flag": "category",
            "PULocationID": "int16",
            "DOLocationID": "int16",
            "payment_type": "int8",
            "fare_amount": "float32",
            "extra": "float32",
            "mta_tax": "float32",
            "tip_amount": "float32",
            "tolls_amount": "float32",
            "improvement_surcharge": "float32",
            "total_amount": "float32",
            "congestion_surcharge": "float32",
            "airport_fee": "float32"
        }

        logger.info(f"Loading taxi data from {file_path}")
        if sample_frac:
            # Sample rows during load to reduce memory for testing
            df = pd.read_csv(
                file_path,
                dtype=dtype_map,
                parse_dates=["tpep_pickup_datetime", "tpep_dropoff_datetime"],
                nrows=int(10_000_000 * sample_frac)
            )
        else:
            df = pd.read_csv(
                file_path,
                dtype=dtype_map,
                parse_dates=["tpep_pickup_datetime", "tpep_dropoff_datetime"]
            )

        # Step 2: Clean invalid data
        logger.info("Cleaning invalid records")
        initial_rows = len(df)
        # Remove trips with negative fares or distances
        df = df[
            (df["fare_amount"] >= 0) &
            (df["trip_distance"] >= 0) &
            (df["passenger_count"] > 0)
        ]
        # Drop rows with missing critical fields
        df = df.dropna(subset=["tpep_pickup_datetime", "total_amount"])
        cleaned_rows = len(df)
        logger.info(f"Removed {initial_rows - cleaned_rows} invalid rows ({((initial_rows - cleaned_rows)/initial_rows)*100:.2f}%)")

        # Step 3: Feature engineering
        logger.info("Engineering features")
        df["pickup_hour"] = df["tpep_pickup_datetime"].dt.hour
        df["pickup_day"] = df["tpep_pickup_datetime"].dt.day_name()
        df["trip_duration_min"] = (df["tpep_dropoff_datetime"] - df["tpep_pickup_datetime"]).dt.total_seconds() / 60

        # Step 4: Aggregate by hour and location
        logger.info("Aggregating data")
        agg_df = df.groupby(["pickup_hour", "PULocationID"]).agg(
            total_trips=("VendorID", "count"),
            avg_fare=("fare_amount", "mean"),
            avg_tip=("tip_amount", "mean"),
            avg_trip_duration=("trip_duration_min", "mean")
        ).reset_index()

        # Step 5: Calculate load and process time
        elapsed_time = time.time() - start_time
        logger.info(f"Processing complete. Total time: {elapsed_time:.2f}s. Output rows: {len(agg_df)}")
        return agg_df

    except FileNotFoundError:
        logger.error(f"File not found: {file_path}")
        raise
    except pd.errors.ParserError as e:
        logger.error(f"CSV parse error: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    # Run full processing for 1.2GB dataset
    try:
        result = process_nyc_taxi_data(sample_frac=None)
        print(f"Top 5 pickup locations by hourly trips:\n{result.sort_values('total_trips', ascending=False).head()}")
    except Exception as e:
        logger.error(f"Failed to process data: {e}")

// Power Query M script to load, clean, and aggregate 2023 NYC Taxi dataset
// Compatible with Power BI Desktop 2.130+
// Author: Senior Data Engineer
// Date: 2024-05-20

// Step 1: Define dataset path and load CSV with error handling
let
    // Try to load the CSV file, handle missing file error
    Source = try Csv.Document(File.Contents("nyc_taxi_2023.csv"), [Delimiter=",", Columns=19, Encoding=1252, QuoteStyle=QuoteStyle.None])
            otherwise error "File not found: nyc_taxi_2023.csv. Please check the path.",

    // Promote first row to headers
    #"Promoted Headers" = Table.PromoteHeaders(Source, [PromoteAllScalars=true]),

    // Step 2: Set explicit data types to match Pandas example
    #"Changed Type" = Table.TransformColumnTypes(#"Promoted Headers", {
        {"VendorID", Int64.Type},
        {"tpep_pickup_datetime", type datetime},
        {"tpep_dropoff_datetime", type datetime},
        {"passenger_count", Int64.Type},
        {"trip_distance", type number},
        {"RatecodeID", Int64.Type},
        {"store_and_fwd_flag", type text},
        {"PULocationID", Int64.Type},
        {"DOLocationID", Int64.Type},
        {"payment_type", Int64.Type},
        {"fare_amount", type number},
        {"extra", type number},
        {"mta_tax", type number},
        {"tip_amount", type number},
        {"tolls_amount", type number},
        {"improvement_surcharge", type number},
        {"total_amount", type number},
        {"congestion_surcharge", type number},
        {"airport_fee", type number}
    }),

    // Step 3: Clean invalid data (match Pandas cleaning logic)
    #"Removed Invalid Rows" = Table.SelectRows(#"Changed Type", each
        [fare_amount] >= 0 and
        [trip_distance] >= 0 and
        [passenger_count] > 0 and
        [tpep_pickup_datetime] <> null and
        [total_amount] <> null
    ),

    // Log row count change (Power BI doesn't have native logging, so add custom column for audit)
    #"Added Row Count" = Table.AddColumn(#"Removed Invalid Rows", "initial_row_count", each Table.RowCount(#"Changed Type")),
    #"Added Cleaned Row Count" = Table.AddColumn(#"Added Row Count", "cleaned_row_count", each Table.RowCount(#"Removed Invalid Rows")),

    // Step 4: Feature engineering (extract hour and day from pickup datetime)
    #"Added Pickup Hour" = Table.AddColumn(#"Added Cleaned Row Count", "pickup_hour", each Time.Hour([tpep_pickup_datetime])),
    #"Added Pickup Day" = Table.AddColumn(#"Added Pickup Hour", "pickup_day", each Date.DayOfWeekName(Date.From([tpep_pickup_datetime]))),
    #"Added Trip Duration" = Table.AddColumn(#"Added Pickup Day", "trip_duration_min", each Duration.TotalMinutes([tpep_dropoff_datetime] - [tpep_pickup_datetime])),

    // Step 5: Aggregate by pickup hour and location (match Pandas groupby)
    #"Grouped Rows" = Table.Group(#"Added Trip Duration", {"pickup_hour", "PULocationID"}, {
        {"total_trips", each Table.RowCount(_), type number},
        {"avg_fare", each List.Average([fare_amount]), type nullable number},
        {"avg_tip", each List.Average([tip_amount]), type nullable number},
        {"avg_trip_duration", each List.Average([trip_duration_min]), type nullable number}
    }),

    // Step 6: Remove audit columns (optional, for production)
    #"Removed Audit Columns" = Table.RemoveColumns(#"Grouped Rows", {"initial_row_count", "cleaned_row_count"})
in
    #"Removed Audit Columns"

import pandas as pd
import pyarrow.parquet as pq
import time
import json
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def benchmark_iot_sensor_processing(
    json_path: str = "iot_sensor_data.json",
    parquet_output: str = "iot_processed.parquet",
    chunk_size: int = 1_000_000
) -> Dict[str, float]:
    """
    Process 12GB IoT Sensor JSON data (120M rows) in chunks, benchmark performance.
    Compatible with Pandas 2.2.1 + PyArrow 16.0.0.

    Args:
        json_path: Path to 12GB JSON file (120M rows, 1 row per sensor reading)
        parquet_output: Path to output Parquet file
        chunk_size: Number of rows per chunk (adjust for memory)

    Returns:
        Dictionary of benchmark metrics (load time, process time, output size)
    """
    metrics = {}
    start_total = time.time()

    try:
        # Step 1: Initialize chunked JSON reader
        logger.info(f"Starting chunked processing of {json_path}")
        chunk_iter = pd.read_json(
            json_path,
            lines=True,  # Newline-delimited JSON
            chunksize=chunk_size,
            dtype={
                "sensor_id": "int32",
                "timestamp": "str",
                "temperature": "float32",
                "humidity": "float32",
                "pressure": "float32",
                "battery_level": "int8"
            }
        )

        # Step 2: Process each chunk, append to Parquet
        chunk_count = 0
        total_rows = 0
        start_load = time.time()

        for chunk in chunk_iter:
            chunk_start = time.time()
            chunk_count += 1

            # Clean chunk: remove invalid sensor readings
            chunk = chunk[
                (chunk["temperature"] > -50) & (chunk["temperature"] < 150) &
                (chunk["humidity"] >= 0) & (chunk["humidity"] <= 100) &
                (chunk["pressure"] > 800) & (chunk["pressure"] < 1200)
            ]

            # Feature engineering: extract hour from timestamp
            chunk["timestamp"] = pd.to_datetime(chunk["timestamp"])
            chunk["hour"] = chunk["timestamp"].dt.hour
            chunk["is_critical"] = (chunk["temperature"] > 90) | (chunk["battery_level"] < 10)

            # Append to Parquet (first chunk writes schema, others append)
            if chunk_count == 1:
                chunk.to_parquet(
                    parquet_output,
                    engine="pyarrow",
                    compression="snappy",
                    index=False
                )
            else:
                chunk.to_parquet(
                    parquet_output,
                    engine="pyarrow",
                    compression="snappy",
                    index=False,
                    append=True  # Requires PyArrow 8.0+
                )

            total_rows += len(chunk)
            chunk_elapsed = time.time() - chunk_start
            logger.info(f"Processed chunk {chunk_count}: {len(chunk)} rows, {chunk_elapsed:.2f}s")

        metrics["load_process_time_s"] = time.time() - start_load
        metrics["total_rows"] = total_rows

        # Step 3: Benchmark Parquet read time (compare to Power BI's 18.7s for 4.7GB)
        start_read = time.time()
        df = pd.read_parquet(parquet_output, engine="pyarrow")
        metrics["parquet_read_time_s"] = time.time() - start_read

        # Step 4: Calculate output file size
        import os
        metrics["output_size_mb"] = os.path.getsize(parquet_output) / (1024 * 1024)

        metrics["total_time_s"] = time.time() - start_total
        logger.info(f"Benchmark complete: {metrics}")
        return metrics

    except FileNotFoundError:
        logger.error(f"JSON file not found: {json_path}")
        raise
    except json.JSONDecodeError as e:
        logger.error(f"Invalid JSON in {json_path}: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    # Run benchmark for 12GB IoT dataset
    try:
        benchmark_results = benchmark_iot_sensor_processing()
        print("IoT Processing Benchmark Results:")
        for k, v in benchmark_results.items():
            print(f"{k}: {v:.2f}")
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")

from sqlalchemy import create_engine

def export_to_sql_for_powerbi(df: pd.DataFrame, table_name: str = "curated_sales"):
    """Export Pandas DataFrame to SQL Server for Power BI consumption"""
    engine = create_engine("mssql+pyodbc://user:pass@sql-server:1433/retail_db?driver=ODBC+Driver+18+for+SQL+Server")
    df.to_sql(table_name, engine, if_exists="replace", index=False)
    print(f"Exported {len(df)} rows to {table_name}")

def downcast_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    """Automatically downcast numeric columns to reduce memory usage"""
    for col in df.select_dtypes(include=["int64"]).columns:
        df[col] = pd.to_numeric(df[col], downcast="integer")
    for col in df.select_dtypes(include=["float64"]).columns:
        df[col] = pd.to_numeric(df[col], downcast="float")
    for col in df.select_dtypes(include=["object"]).columns:
        num_unique = df[col].nunique()
        if num_unique / len(df) < 0.5:  # Low cardinality
            df[col] = df[col].astype("category")
        else:
            df[col] = df[col].astype("string")
    return df

import requests

def trigger_powerbi_refresh(workspace_id: str, dataset_id: str, token: str):
    """Trigger Power BI dataset refresh via REST API"""
    url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/refreshes"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    response = requests.post(url, headers=headers, json={"notifyOption": "NoNotification"})
    if response.status_code == 202:
        print("Refresh triggered successfully")
    else:
        print(f"Refresh failed: {response.status_code} {response.text}")