Queries That Save Time Tableau vs Pandas: What You Need to Know

import pandas as pd
import numpy as np
import time
from typing import Optional

def run_pandas_filtered_agg(
    parquet_path: str,
    filter_conditions: list[dict],
    group_cols: list[str],
    agg_cols: dict[str, str],
    max_memory_gb: Optional[float] = 32.0
) -> pd.DataFrame:
    """
    Execute filtered aggregation on Parquet dataset using Pandas 2.2.1.

    Args:
        parquet_path: Path to Parquet file
        filter_conditions: List of dicts with keys 'col', 'op', 'val' (e.g., {'col': 'fare', 'op': 'gt', 'val': 10.0})
        group_cols: Columns to group by
        agg_cols: Dict mapping column name to aggregation function (e.g., {'total_fare': 'sum'})
        max_memory_gb: Max memory allowed for operation, raises MemoryError if exceeded

    Returns:
        Aggregated DataFrame
    """
    start_time = time.time()
    try:
        # Check if file exists
        import os
        if not os.path.exists(parquet_path):
            raise FileNotFoundError(f"Parquet file not found at {parquet_path}")

        # Read Parquet with memory limit check (simplified)
        df = pd.read_parquet(parquet_path)
        est_memory_mb = df.memory_usage(deep=True).sum() / (1024 * 1024)
        if est_memory_mb / 1024 > max_memory_gb:
            raise MemoryError(f"Dataset estimated to use {est_memory_mb:.2f}MB, exceeds {max_memory_gb}GB limit")

        # Apply filter conditions
        mask = pd.Series(True, index=df.index)
        for cond in filter_conditions:
            col = cond['col']
            op = cond['op']
            val = cond['val']
            if col not in df.columns:
                raise ValueError(f"Filter column {col} not found in dataset")
            if op == 'gt':
                mask &= (df[col] > val)
            elif op == 'lt':
                mask &= (df[col] < val)
            elif op == 'eq':
                mask &= (df[col] == val)
            else:
                raise ValueError(f"Unsupported filter operator {op}")

        filtered_df = df[mask].copy()
        print(f"Filtered dataset to {len(filtered_df)} rows ({len(filtered_df)/len(df):.2%} of original)")

        # Validate group and agg columns
        for col in group_cols:
            if col not in filtered_df.columns:
                raise ValueError(f"Group column {col} not found in filtered dataset")
        for col in agg_cols.keys():
            if col not in filtered_df.columns:
                raise ValueError(f"Aggregation column {col} not found in filtered dataset")

        # Execute aggregation
        agg_result = filtered_df.groupby(group_cols).agg(agg_cols).reset_index()

        elapsed = time.time() - start_time
        print(f"Pandas aggregation completed in {elapsed:.2f}s")
        return agg_result

    except FileNotFoundError as e:
        print(f"File error: {e}")
        raise
    except MemoryError as e:
        print(f"Memory error: {e}")
        raise
    except ValueError as e:
        print(f"Value error: {e}")
        raise
    except Exception as e:
        print(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    # Example usage with NYC Taxi 2023 dataset (10M rows)
    try:
        result = run_pandas_filtered_agg(
            parquet_path="nyc_taxi_2023.parquet",
            filter_conditions=[
                {"col": "fare_amount", "op": "gt", "val": 10.0},
                {"col": "trip_distance", "op": "gt", "val": 2.0}
            ],
            group_cols=["vendor_id", "payment_type"],
            agg_cols={
                "fare_amount": "sum",
                "tip_amount": "mean",
                "trip_distance": "count"
            }
        )
        print(f"Aggregation result shape: {result.shape}")
        print(result.head())
    except Exception as e:
        print(f"Failed to run aggregation: {e}")

import tableauserverclient as tsc  # https://github.com/tableau/server-client-python
import os
import time
from typing import Optional, Dict, Any

def run_tableau_datasource_query(
    server_url: str,
    site_id: str,
    token_name: str,
    token_secret: str,
    datasource_name: str,
    query_params: Optional[Dict[str, Any]] = None
) -> list[dict]:
    """
    Execute a query against a published Tableau data source using Tableau Server Client 0.24.

    Args:
        server_url: Tableau Server URL (e.g., https://tableau.example.com)
        site_id: Tableau Site ID (use "" for default site)
        token_name: Personal Access Token name
        token_secret: Personal Access Token secret
        datasource_name: Name of published data source to query
        query_params: Optional parameters to pass to the query (e.g., filters)

    Returns:
        List of dicts representing query results
    """
    start_time = time.time()
    query_params = query_params or {}

    try:
        # Validate inputs
        if not all([server_url, token_name, token_secret, datasource_name]):
            raise ValueError("All authentication and datasource parameters are required")

        # Initialize Tableau Auth and Server
        auth = tsc.PersonalAccessTokenAuth(
            token_name=token_name,
            personal_access_token=token_secret,
            site_id=site_id
        )
        server = tsc.Server(server_url, use_server_version=True)

        # Sign in to Tableau Server
        with server.auth.sign_in(auth):
            print(f"Signed in to Tableau Server at {server_url}")

            # Fetch target data source
            all_datasources = list(tsc.Pager(server.datasources))
            target_datasource = None
            for ds in all_datasources:
                if ds.name == datasource_name:
                    target_datasource = ds
                    break

            if not target_datasource:
                raise ValueError(f"Datasource {datasource_name} not found on server")

            print(f"Found datasource: {target_datasource.name} (ID: {target_datasource.id})")

            # Query data source (note: Tableau's REST API has limited query capabilities,
            # this uses the experimental query endpoint available in Tableau 2024.2+)
            if not hasattr(server.datasources, 'query_data'):
                raise NotImplementedError("Datasource query not supported in this Tableau Server version")

            # Apply query parameters (filters)
            filter_expr = ""
            if query_params.get("filters"):
                filter_parts = []
                for f in query_params["filters"]:
                    filter_parts.append(f"{f['col']} {f['op']} {f['val']}")
                filter_expr = " AND ".join(filter_parts)

            # Execute query
            query_result = server.datasources.query_data(
                datasource=target_datasource,
                max_rows=query_params.get("max_rows", 10000),
                filter=filter_expr
            )

            # Convert to list of dicts
            results = []
            for row in query_result:
                results.append(dict(row))

            elapsed = time.time() - start_time
            print(f"Tableau datasource query completed in {elapsed:.2f}s, returned {len(results)} rows")
            return results

    except tsc.server.exceptions.ServerLoginError as e:
        print(f"Tableau login failed: {e}")
        raise
    except tsc.server.exceptions.ServerError as e:
        print(f"Tableau server error: {e}")
        raise
    except ValueError as e:
        print(f"Value error: {e}")
        raise
    except NotImplementedError as e:
        print(f"Not implemented: {e}")
        raise
    except Exception as e:
        print(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    # Example usage (replace with your Tableau Server credentials)
    try:
        # Load credentials from environment variables for security
        server_url = os.getenv("TABLEAU_SERVER_URL")
        token_name = os.getenv("TABLEAU_TOKEN_NAME")
        token_secret = os.getenv("TABLEAU_TOKEN_SECRET")
        site_id = os.getenv("TABLEAU_SITE_ID", "")

        if not all([server_url, token_name, token_secret]):
            raise ValueError("Missing required environment variables for Tableau auth")

        results = run_tableau_datasource_query(
            server_url=server_url,
            site_id=site_id,
            token_name=token_name,
            token_secret=token_secret,
            datasource_name="NYC Taxi 2023 Hyper Extract",
            query_params={
                "filters": [
                    {"col": "fare_amount", "op": ">", "val": 10},
                    {"col": "trip_distance", "op": ">", "val": 2}
                ],
                "max_rows": 10000
            }
        )
        print(f"First 3 results: {results[:3]}")
    except Exception as e:
        print(f"Failed to run Tableau query: {e}")

import duckdb
import pandas as pd
import tableauserverclient as tsc  # https://github.com/tableau/server-client-python
import os
import time
from typing import Tuple, Optional

def run_hybrid_shared_sql_query(
    sql_query: str,
    parquet_path: str,
    tableau_datasource_name: Optional[str] = None,
    tableau_server_config: Optional[dict] = None
) -> Tuple[pd.DataFrame, Optional[list[dict]]]:
    """
    Execute a shared SQL query across Pandas (via DuckDB) and Tableau, enabling consistent results.

    Args:
        sql_query: Shared SQL query to execute (must be DuckDB-compatible)
        parquet_path: Path to Parquet dataset for DuckDB/Pandas execution
        tableau_datasource_name: If provided, publish query result to Tableau as a data source
        tableau_server_config: Dict with Tableau server config (server_url, site_id, token_name, token_secret)

    Returns:
        Tuple of (Pandas DataFrame result, Tableau query result if applicable)
    """
    start_time = time.time()
    pandas_result = None
    tableau_result = None

    try:
        # --- Step 1: Execute query in Pandas via DuckDB (https://github.com/duckdb/duckdb) ---
        print("Executing shared SQL query in Pandas via DuckDB...")
        if not os.path.exists(parquet_path):
            raise FileNotFoundError(f"Parquet file not found at {parquet_path}")

        # Connect to DuckDB, register Parquet file as a table
        con = duckdb.connect(database=":memory:")
        con.execute(f"CREATE VIEW taxi_data AS SELECT * FROM parquet_scan('{parquet_path}')")

        # Execute shared SQL query
        pandas_result = con.execute(sql_query).fetchdf()
        print(f"Pandas (DuckDB) result: {len(pandas_result)} rows, columns: {list(pandas_result.columns)}")

        # --- Step 2: Execute same query in Tableau if config provided ---
        if tableau_datasource_name and tableau_server_config:
            print(f"Publishing shared query result to Tableau as {tableau_datasource_name}...")
            # Initialize Tableau auth
            auth = tsc.PersonalAccessTokenAuth(
                token_name=tableau_server_config["token_name"],
                personal_access_token=tableau_server_config["token_secret"],
                site_id=tableau_server_config.get("site_id", "")
            )
            server = tsc.Server(tableau_server_config["server_url"], use_server_version=True)

            with server.auth.sign_in(auth):
                # Create a Hyper extract from the query result to publish to Tableau
                import pyhyper  # https://github.com/tableau/hyper-api-python
                hyper_path = f"{tableau_datasource_name.replace(' ', '_')}.hyper"

                # Write Pandas result to Hyper file
                with pyhyper.HyperProcess(telemetry=pyhyper.Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as hyper:
                    with pyhyper.Connection(hyper.endpoint, hyper_path, pyhyper.CreateMode.CREATE_IF_NOT_EXISTS) as connection:
                        # Create table schema from Pandas DataFrame
                        connection.catalog.create_table_from_dataframe(
                            table_name="shared_query_result",
                            dataframe=pandas_result,
                            connection=connection
                        )

                # Publish Hyper file to Tableau Server as a data source
                datasource = tsc.DatasourceItem(tableau_datasource_name)
                datasource = server.datasources.publish(
                    datasource,
                    hyper_path,
                    mode=tsc.Server.PublishMode.Overwrite
                )
                print(f"Published data source to Tableau: {datasource.name} (ID: {datasource.id})")

                # Query the published data source to verify
                if hasattr(server.datasources, 'query_data'):
                    tableau_result = list(server.datasources.query_data(datasource, max_rows=10000))
                    print(f"Tableau query result: {len(tableau_result)} rows")

                # Clean up local Hyper file
                if os.path.exists(hyper_path):
                    os.remove(hyper_path)

        elapsed = time.time() - start_time
        print(f"Hybrid query workflow completed in {elapsed:.2f}s")
        return pandas_result, tableau_result

    except FileNotFoundError as e:
        print(f"File error: {e}")
        raise
    except duckdb.Error as e:
        print(f"DuckDB error: {e}")
        raise
    except tsc.server.exceptions.ServerError as e:
        print(f"Tableau error: {e}")
        raise
    except Exception as e:
        print(f"Unexpected error: {e}")
        raise

if __name__ == "__main__":
    # Shared SQL query (works in DuckDB and can be used as Custom SQL in Tableau)
    shared_sql = """
    SELECT 
        vendor_id,
        payment_type,
        SUM(fare_amount) AS total_fare,
        AVG(tip_amount) AS avg_tip,
        COUNT(*) AS trip_count
    FROM taxi_data
    WHERE fare_amount > 10 AND trip_distance > 2
    GROUP BY vendor_id, payment_type
    ORDER BY total_fare DESC
    """

    try:
        # Run without Tableau publishing first
        pandas_result, _ = run_hybrid_shared_sql_query(
            sql_query=shared_sql,
            parquet_path="nyc_taxi_2023.parquet"
        )
        print("Pandas shared SQL result head:")
        print(pandas_result.head())

        # Uncomment below to run with Tableau publishing (requires env vars)
        # tableau_config = {
        #     "server_url": os.getenv("TABLEAU_SERVER_URL"),
        #     "site_id": os.getenv("TABLEAU_SITE_ID", ""),
        #     "token_name": os.getenv("TABLEAU_TOKEN_NAME"),
        #     "token_secret": os.getenv("TABLEAU_TOKEN_SECRET")
        # }
        # if all(tableau_config.values()):
        #     run_hybrid_shared_sql_query(
        #         sql_query=shared_sql,
        #         parquet_path="nyc_taxi_2023.parquet",
        #         tableau_datasource_name="Shared SQL Taxi Query",
        #         tableau_server_config=tableau_config
        #     )
    except Exception as e:
        print(f"Hybrid workflow failed: {e}")

shared_sql = """
SELECT 
    merchant_category,
    DATE_TRUNC('day', transaction_time) AS txn_day,
    SUM(transaction_amount) AS daily_volume,
    COUNT(*) AS txn_count
FROM transaction_data
WHERE transaction_status = 'approved'
GROUP BY merchant_category, txn_day
ORDER BY daily_volume DESC
"""

import cProfile
import pstats
import pandas as pd

def profile_pandas_query(parquet_path: str):
    def run_query():
        df = pd.read_parquet(parquet_path)
        return df[df["fare_amount"] > 10].groupby("vendor_id").sum()

    profiler = cProfile.Profile()
    profiler.enable()
    run_query()
    profiler.disable()

    stats = pstats.Stats(profiler)
    stats.sort_stats("cumtime")
    stats.print_stats(10)

profile_pandas_query("nyc_taxi_2023.parquet")

import pandas as pd
import pyhyper  # https://github.com/tableau/hyper-api-python

def export_to_tableau_hyper(df: pd.DataFrame, hyper_path: str, table_name: str):
    with pyhyper.HyperProcess(telemetry=pyhyper.Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as hyper:
        with pyhyper.Connection(hyper.endpoint, hyper_path, pyhyper.CreateMode.CREATE_IF_NOT_EXISTS) as connection:
            connection.catalog.create_table_from_dataframe(
                table_name=table_name,
                dataframe=df,
                connection=connection
            )
    print(f"Exported {len(df)} rows to {hyper_path}")

# Example usage
df = pd.read_parquet("exploratory_result.parquet")
export_to_tableau_hyper(df, "exploratory_result.hyper", "exploratory_data")