49 - BigQuery Advanced Optimization

from google.cloud import bigquery  # Import BigQuery client library
from google.cloud.bigquery import SchemaField  # Import SchemaField for table schema
from typing import List  # Import List for type annotations

# Define schema with date for partitioning
schema: List[SchemaField] = [
    SchemaField("product", "STRING"),  # Product name column (string)
    SchemaField("price", "FLOAT"),  # Price column (float)
    SchemaField("quantity", "INTEGER"),  # Quantity column (integer)
    SchemaField("sale_date", "DATE")  # Partitioning column (date)
]

# Create partitioned table
client = bigquery.Client()  # Initialize BigQuery client for API interactions
table_id = "your_project.your_dataset.sales_partitioned"  # Define table ID
table = bigquery.Table(table_id, schema=schema)  # Create table object with schema
table.time_partitioning = bigquery.TimePartitioning(
    type_=bigquery.TimePartitioningType.DAY,  # Set daily partitioning
    field="sale_date"  # Partition by sale_date column
)
table = client.create_table(table)  # Create partitioned table in BigQuery
print(f"Created partitioned table {table_id}")  # Confirm table creation

# Create clustered table
table_id = "your_project.your_dataset.sales_clustered"  # Define table ID for clustered table
table = bigquery.Table(table_id, schema=schema)  # Create table object with same schema
table.clustering_fields = ["product"]  # Cluster by product column for efficient filtering
table = client.create_table(table)  # Create clustered table in BigQuery
print(f"Created clustered table {table_id}")  # Confirm table creation

# Example query benefiting from clustering
query = """
SELECT product, SUM(price * quantity) as total_sales
FROM your_project.your_dataset.sales_clustered
WHERE product = 'Halal Laptop'
GROUP BY product
"""  # Clustering reduces scanned data for product-specific queries

# Create materialized view
query = """
CREATE MATERIALIZED VIEW your_project.your_dataset.sales_mv
AS
SELECT product, SUM(price * quantity) as total_sales
FROM your_project.your_dataset.sales_partitioned
GROUP BY product
"""  # Define SQL query to create materialized view for total sales by product
client.query(query).result()  # Execute query and wait for completion
print("Created materialized view sales_mv")  # Confirm view creation

from google.cloud import bigquery  # Import BigQuery client library
from pydantic import BaseModel  # Import Pydantic for data validation
from typing import Dict, Any, List  # Import types for annotations

# Pydantic model ensures type safety for BigQuery data, validated by Pyright
class SalesRecord(BaseModel):
    product: str  # Product name (string)
    price: float  # Price (float)
    quantity: int  # Quantity (integer)
    sale_date: str  # ISO date string (e.g., "2023-10-01")

def load_sales(client: bigquery.Client, table_id: str, records: List[Dict[str, Any]]) -> None:
    """Load sales records into BigQuery."""
    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_APPEND",  # Append data to existing table
        schema=[
            bigquery.SchemaField("product", "STRING"),  # Define schema: product column
            bigquery.SchemaField("price", "FLOAT"),  # Define schema: price column
            bigquery.SchemaField("quantity", "INTEGER"),  # Define schema: quantity column
            bigquery.SchemaField("sale_date", "DATE")  # Define schema: sale_date column
        ]
    )  # Configure load job with schema and write disposition
    job = client.load_table_from_json(records, table_id, job_config=job_config)  # Load JSON records
    job.result()  # Wait for job completion
    print(f"Loaded {len(records)} records to {table_id}")  # Confirm records loaded

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/generate_sales_dates.py
import pandas as pd  # Import Pandas for CSV manipulation
from datetime import datetime, timedelta  # Import datetime for date generation

def generate_sales_with_dates(input_path: str, output_path: str) -> None:
    """Generate sales_with_dates.csv with sale_date column."""
    df = pd.read_csv(input_path)  # Load input CSV into DataFrame
    # Generate dates starting from 2023-10-01
    start_date = datetime(2023, 10, 1)  # Set start date
    df["sale_date"] = [start_date + timedelta(days=i % 5) for i in range(len(df))]  # Assign dates (cycle every 5 days)
    df["sale_date"] = df["sale_date"].dt.strftime("%Y-%m-%d")  # Format dates as ISO strings
    df.to_csv(output_path, index=False)  # Save DataFrame to output CSV
    print(f"Generated {output_path}")  # Confirm file creation

# Run generation if script is executed directly
if __name__ == "__main__":
    generate_sales_with_dates("data/sales.csv", "data/sales_with_dates.csv")  # Generate CSV

# File: de-onboarding/utils.py (extended from Chapter 3)
import yaml  # Import PyYAML for configuration parsing
from typing import Dict, Any  # Import types for annotations

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    print(f"Opening config: {config_path}")  # Log config file path
    with open(config_path, "r") as file:  # Open YAML file
        config = yaml.safe_load(file)  # Parse YAML into dictionary
    print(f"Loaded config: {config}")  # Log loaded configuration
    return config  # Return configuration dictionary

# File: de-onboarding/optimize_bigquery.py
from google.cloud import bigquery  # Import BigQuery client library
from google.cloud.bigquery import SchemaField  # Import SchemaField for table schema
import pandas as pd  # Import Pandas for CSV processing
from pydantic import BaseModel, ValidationError  # Import Pydantic for validation
from typing import List, Dict, Any  # Import types for annotations
import json  # Import JSON for output
import os  # Import os for file operations
from generate_sales_dates import generate_sales_with_dates  # Import dataset generation script
import utils  # Import utilities module

# Pydantic model ensures type safety for BigQuery data, validated by Pyright
class SalesRecord(BaseModel):
    product: str  # Product name (string)
    price: float  # Price (float)
    quantity: int  # Quantity (integer)
    sale_date: str  # ISO date string (e.g., "2023-10-01")

def create_partitioned_table(client: bigquery.Client, table_id: str) -> None:
    """Create a partitioned BigQuery table."""
    schema: List[SchemaField] = [
        SchemaField("product", "STRING"),  # Product column
        SchemaField("price", "FLOAT"),  # Price column
        SchemaField("quantity", "INTEGER"),  # Quantity column
        SchemaField("sale_date", "DATE")  # Date column for partitioning
    ]  # Define table schema
    table = bigquery.Table(table_id, schema=schema)  # Create table object
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,  # Set daily partitioning
        field="sale_date"  # Partition by sale_date
    )  # Configure time-based partitioning
    client.create_table(table)  # Create table in BigQuery
    print(f"Created partitioned table {table_id}")  # Confirm creation

def create_clustered_table(client: bigquery.Client, table_id: str) -> None:
    """Create a clustered BigQuery table."""
    schema: List[SchemaField] = [
        SchemaField("product", "STRING"),  # Product column
        SchemaField("price", "FLOAT"),  # Price column
        SchemaField("quantity", "INTEGER"),  # Quantity column
        SchemaField("sale_date", "DATE")  # Date column
    ]  # Define table schema
    table = bigquery.Table(table_id, schema=schema)  # Create table object
    table.clustering_fields = ["product"]  # Cluster by product for efficient filtering
    client.create_table(table)  # Create table in BigQuery
    print(f"Created clustered table {table_id}")  # Confirm creation

def create_materialized_view(client: bigquery.Client, view_id: str, source_table: str) -> None:
    """Create a materialized view for total sales."""
    query = f"""
    CREATE MATERIALIZED VIEW {view_id}
    AS
    SELECT product, SUM(price * quantity) as total_sales
    FROM {source_table}
    GROUP BY product
    """  # Define SQL to create materialized view aggregating sales by product
    client.query(query).result()  # Execute query and wait for completion
    print(f"Created materialized view {view_id}")  # Confirm view creation

def load_sales_data(client: bigquery.Client, table_id: str, csv_path: str, config: Dict[str, Any]) -> tuple[List[Dict[str, Any]], int]:
    """Load and validate sales data into BigQuery."""
    df = pd.read_csv(csv_path, encoding='utf-8')  # Load CSV into DataFrame with UTF-8 encoding
    print("Initial DataFrame:")  # Log DataFrame preview
    print(df.head())  # Display first 5 rows for debugging

    records: List[Dict[str, Any]] = []  # Initialize list for valid records
    invalid_count = 0  # Initialize counter for invalid records
    for _, row in df.iterrows():  # Iterate through DataFrame rows
        try:
            price = pd.to_numeric(row["price"], errors="coerce")  # Convert price to numeric, NaN if invalid
            if pd.isna(price):  # Check for invalid price
                raise ValueError("Invalid price")
            record = {
                "product": str(row["product"]) if pd.notna(row["product"]) else "",  # Handle missing product
                "price": float(price),  # Convert price to float
                "quantity": int(row["quantity"]) if pd.notna(row["quantity"]) else 0,  # Handle missing quantity
                "sale_date": str(row["sale_date"])  # Use sale_date from CSV
            }  # Create record dictionary
            sales_record = SalesRecord(**record)  # Validate record with Pydantic
            if (sales_record.product.startswith(config["product_prefix"]) and  # Validate product prefix
                sales_record.price >= config["min_price"] and  # Validate minimum price
                sales_record.quantity <= config["max_quantity"] and  # Validate maximum quantity
                sales_record.price > 0):  # Ensure positive price
                records.append(sales_record.dict())  # Add valid record to list
            else:
                invalid_count += 1  # Increment invalid counter
                print(f"Invalid record: {record}")  # Log invalid record
        except (ValidationError, ValueError) as e:  # Handle validation or type errors
            invalid_count += 1  # Increment invalid counter
            print(f"Validation error: {e}, Record: {row}, Types: {row.dtypes}")  # Log error details

    if records:  # Check if there are valid records to load
        job_config = bigquery.LoadJobConfig(
            write_disposition="WRITE_APPEND",  # Append data to table
            schema=[
                SchemaField("product", "STRING"),  # Schema: product
                SchemaField("price", "FLOAT"),  # Schema: price
                SchemaField("quantity", "INTEGER"),  # Schema: quantity
                SchemaField("sale_date", "DATE")  # Schema: sale_date
            ]
        )  # Configure load job
        job = client.load_table_from_json(records, table_id, job_config=job_config)  # Load records
        job.result()  # Wait for completion
        print(f"Loaded {len(records)} records to {table_id}")  # Confirm records loaded

    return records, invalid_count  # Return valid records and invalid count

def run_optimized_query(client: bigquery.Client, table_id: str) -> Dict[str, Any]:
    """Run an optimized query for recent sales."""
    query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    WHERE sale_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
    GROUP BY product
    """  # Define optimized SQL query using partitioning
    job_config = bigquery.QueryJobConfig(
        query_parameters=[],  # No parameters needed
        use_query_cache=True  # Enable caching to reuse query results, reducing costs
    )  # Configure query job; leverage Chapter 27’s SQL techniques for optimized filtering
    query_job = client.query(query, job_config=job_config)  # Execute query
    results = query_job.result()  # Get query results
    total_bytes = query_job.total_bytes_processed  # Get bytes scanned
    print(f"Query scanned {total_bytes} bytes")  # Log bytes scanned

    sales_data = {row.product: float(row.total_sales) for row in results}  # Convert results to dictionary
    return {
        "sales_data": sales_data,  # Query results
        "bytes_processed": total_bytes  # Bytes scanned
    }  # Return query results and metrics

def calculate_query_cost(bytes_processed: int) -> float:
    """Calculate query cost based on bytes processed ($6.25/TB)."""
    terabytes = bytes_processed / 1e12  # Convert bytes to terabytes
    cost = terabytes * 6.25  # Calculate cost at $6.25 per TB
    return round(cost, 4)  # Round to 4 decimal places

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export optimization results to JSON."""
    print(f"Writing to: {json_path}")  # Log output path
    with open(json_path, "w") as file:  # Open JSON file for writing
        json.dump(results, file, indent=2)  # Write results with indentation
    print(f"Exported results to {json_path}")  # Confirm export

def main() -> None:
    """Main function to optimize BigQuery sales warehouse."""
    project_id = "your_project"  # Define Google Cloud project ID
    dataset_id = "your_dataset"  # Define BigQuery dataset ID
    input_csv_path = "data/sales.csv"  # Input CSV path
    csv_path = "data/sales_with_dates.csv"  # Generated CSV path
    config_path = "data/config.yaml"  # Config YAML path
    json_path = "data/optimization_report.json"  # Output JSON path

    # Generate sales_with_dates.csv with sale_date column
    generate_sales_with_dates(input_csv_path, csv_path)  # Run dataset generation

    client = bigquery.Client(project=project_id)  # Initialize BigQuery client
    config = utils.read_config(config_path)  # Load configuration

    # Create tables
    partitioned_table = f"{project_id}.{dataset_id}.sales_partitioned"  # Partitioned table ID
    clustered_table = f"{project_id}.{dataset_id}.sales_clustered"  # Clustered table ID
    view_id = f"{project_id}.{dataset_id}.sales_mv"  # Materialized view ID

    create_partitioned_table(client, partitioned_table)  # Create partitioned table
    create_clustered_table(client, clustered_table)  # Create clustered table

    # Load data into partitioned table
    records, invalid_count = load_sales_data(client, partitioned_table, csv_path, config)  # Load and validate data
    if records:  # Check if valid records exist
        load_sales_data(client, clustered_table, csv_path, config)  # Load into clustered table
        create_materialized_view(client, view_id, partitioned_table)  # Create materialized view

    # Run optimized query on clustered table
    query_results = run_optimized_query(client, clustered_table)  # Execute query

    # Calculate query cost
    query_cost = calculate_query_cost(query_results["bytes_processed"])  # Compute cost

    # Export results
    results = {
        "valid_records": len(records),  # Number of valid records
        "invalid_records": invalid_count,  # Number of invalid records
        "query_results": query_results["sales_data"],  # Query results
        "bytes_processed": query_results["bytes_processed"],  # Bytes scanned
        "query_cost_usd": query_cost  # Query cost in USD
    }  # Compile results dictionary
    export_results(results, json_path)  # Export to JSON

    # Print optimization report
    print("\nOptimization Report:")  # Log report header
    print(f"Valid Records: {results['valid_records']}")  # Log valid record count
    print(f"Invalid Records: {results['invalid_records']}")  # Log invalid record count
    print(f"Query Results: {results['query_results']}")  # Log query results
    print(f"Bytes Processed: {results['bytes_processed']}")  # Log bytes scanned
    print(f"Query Cost: ${results['query_cost_usd']}")  # Log query cost

# Run main function if script is executed directly
if __name__ == "__main__":
    main()  # Execute main workflow

# File: de-onboarding/tests/test_optimize_bigquery.py
from google.cloud import bigquery  # Import BigQuery client library
import pytest  # Import pytest for testing
import os  # Import os for file operations
from optimize_bigquery import load_sales_data, run_optimized_query, SalesRecord, calculate_query_cost  # Import functions to test
from typing import Dict, Any  # Import types for annotations

@pytest.fixture
def client():
    """Fixture to provide BigQuery client."""
    return bigquery.Client(project="your_project")  # Initialize client with project ID

@pytest.fixture
def config():
    """Fixture to provide configuration dictionary."""
    return {
        "min_price": 10.0,  # Minimum price for validation
        "max_quantity": 100,  # Maximum quantity for validation
        "product_prefix": "Halal",  # Required product prefix
        "required_fields": ["product", "price", "quantity"],  # Required CSV fields
        "max_decimals": 2  # Maximum decimal places for price
    }  # Return config dictionary

def test_load_sales_data(client: bigquery.Client, config: Dict[str, Any]):
    """Test loading and validating sales data."""
    table_id = "your_project.your_dataset.test_sales"  # Define test table ID
    records, invalid_count = load_sales_data(client, table_id, "data/sales_with_dates.csv", config)  # Load data
    assert len(records) == 3, "Should load 3 valid records"  # Verify valid record count
    assert invalid_count == 3, "Should detect 3 invalid records"  # Verify invalid record count
    for record in records:  # Check each valid record
        assert record["product"].startswith("Halal")  # Verify product prefix
        assert record["price"] >= config["min_price"]  # Verify minimum price
        assert record["quantity"] <= config["max_quantity"]  # Verify maximum quantity

def test_optimized_query(client: bigquery.Client):
    """Test optimized query execution."""
    table_id = "your_project.your_dataset.sales_clustered"  # Define clustered table ID
    results = run_optimized_query(client, table_id)  # Run query
    assert "sales_data" in results  # Verify results contain sales data
    assert "bytes_processed" in results  # Verify bytes processed included
    assert results["bytes_processed"] > 0  # Verify non-zero bytes processed

def test_query_cost():
    """Test query cost calculation."""
    bytes_processed = 1000000000  # Set test value: 1GB
    cost = calculate_query_cost(bytes_processed)  # Calculate cost
    assert cost == 0.0063, "Should calculate cost for 1GB at $6.25/TB"  # Verify cost

{
  "valid_records": 3,
  "invalid_records": 3,
  "query_results": {
    "Halal Laptop": 1999.98,
    "Halal Mouse": 249.9,
    "Halal Keyboard": 249.95
  },
  "bytes_processed": 123456,
  "query_cost_usd": 0.0008
}

SELECT product, SUM(price * quantity) as total_sales
FROM your_project.your_dataset.sales_clustered
WHERE product = 'Halal Laptop'
GROUP BY product

def run_query(client: bigquery.Client, table_id: str) -> Dict[str, Any]:
    query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    GROUP BY product
    """
    query_job = client.query(query)
    results = query_job.result()
    return {row.product: float(row.total_sales) for row in results}

def compare_query_cost(client: bigquery.Client, table_id: str) -> Dict[str, int]:
    # Unoptimized query
    unopt_query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    GROUP BY product
    """
    unopt_job = client.query(unopt_query)
    unopt_bytes = unopt_job.result().total_bytes_processed

    # Optimized query
    opt_query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    WHERE sale_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
    GROUP BY product
    """
    opt_job = client.query(opt_query, job_config=bigquery.QueryJobConfig(use_query_cache=True))
    opt_bytes = opt_job.result().total_bytes_processed

    return {"unoptimized_bytes": unopt_bytes, "optimized_bytes": opt_bytes}

from google.cloud import bigquery  # Import BigQuery client library
from google.cloud.bigquery import SchemaField  # Import SchemaField for schema
from typing import List  # Import List for type annotations

def create_partitioned_table(client: bigquery.Client, table_id: str) -> None:
    """Create a partitioned BigQuery table."""
    schema: List[SchemaField] = [
        SchemaField("product", "STRING"),  # Product column
        SchemaField("price", "FLOAT"),  # Price column
        SchemaField("quantity", "INTEGER"),  # Quantity column
        SchemaField("sale_date", "DATE")  # Date column for partitioning
    ]  # Define table schema
    table = bigquery.Table(table_id, schema=schema)  # Create table object
    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,  # Set daily partitioning
        field="sale_date"  # Partition by sale_date
    )  # Configure partitioning
    client.create_table(table)  # Create table in BigQuery
    print(f"Created partitioned table {table_id}")  # Confirm creation

from google.cloud import bigquery  # Import BigQuery client library
from typing import Dict, Any  # Import types for annotations

def query_clustered_table(client: bigquery.Client, table_id: str) -> Dict[str, Any]:
    """Query clustered table for Halal products."""
    query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    WHERE product LIKE 'Halal%'
    GROUP BY product
    """  # Define SQL query to aggregate sales for Halal products
    query_job = client.query(query, job_config=bigquery.QueryJobConfig(use_query_cache=True))  # Execute query with caching
    results = query_job.result()  # Get query results
    print(f"Query scanned {query_job.total_bytes_processed} bytes")  # Log bytes scanned
    return {row.product: float(row.total_sales) for row in results}  # Convert results to dictionary

def analyze_clustering_benefits() -> str:
    """Explain clustering benefits for a product-based query."""
    explanation = (
        "Clustering co-locates 'Halal Laptop' rows, scanning 500MB instead of 1GB, "
        "as only relevant data blocks are accessed."
    )  # Define explanation of clustering benefits
    print(explanation)  # Print explanation
    return explanation  # Return explanation string

# Test
analyze_clustering_benefits()  # Run analysis

from google.cloud import bigquery  # Import BigQuery client library
from typing import Dict, Any  # Import types for annotations

def run_query(client: bigquery.Client, table_id: str) -> Dict[str, Any]:
    """Run optimized query with partitioning."""
    query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    WHERE sale_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
    GROUP BY product
    """  # Define optimized SQL query using partitioning
    job_config = bigquery.QueryJobConfig(use_query_cache=True)  # Enable query caching
    query_job = client.query(query, job_config=job_config)  # Execute query
    results = query_job.result()  # Get query results
    print(f"Query scanned {query_job.total_bytes_processed} bytes")  # Log bytes scanned
    return {row.product: float(row.total_sales) for row in results}  # Convert results to dictionary

from google.cloud import bigquery  # Import BigQuery client library
from typing import Dict, Any  # Import types for annotations

def query_materialized_view(client: bigquery.Client, view_id: str) -> Dict[str, Any]:
    """Query materialized view for total sales by product."""
    query = f"""
    SELECT product, total_sales
    FROM {view_id}
    """  # Define SQL query to retrieve pre-computed sales from materialized view
    query_job = client.query(query, job_config=bigquery.QueryJobConfig(use_query_cache=True))  # Execute query with caching
    results = query_job.result()  # Get query results
    print(f"Query scanned {query_job.total_bytes_processed} bytes")  # Log bytes scanned
    return {row.product: float(row.total_sales) for row in results}  # Convert results to dictionary

from google.cloud import bigquery  # Import BigQuery client library
from typing import Dict, int  # Import types for annotations

def compare_query_cost(client: bigquery.Client, table_id: str) -> Dict[str, int]:
    """Compare bytes scanned by unoptimized vs. optimized queries."""
    # Unoptimized query: scans entire table
    unopt_query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    GROUP BY product
    """  # Define unoptimized SQL query
    unopt_job = client.query(unopt_query)  # Execute unoptimized query
    unopt_bytes = unopt_job.result().total_bytes_processed  # Get bytes scanned

    # Optimized query: uses partitioning
    opt_query = f"""
    SELECT product, SUM(price * quantity) as total_sales
    FROM {table_id}
    WHERE sale_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
    GROUP BY product
    """  # Define optimized SQL query
    opt_job = client.query(opt_query, job_config=bigquery.QueryJobConfig(use_query_cache=True))  # Execute with caching
    opt_bytes = opt_job.result().total_bytes_processed  # Get bytes scanned

    print(f"Unoptimized: {unopt_bytes} bytes, Optimized: {opt_bytes} bytes")  # Log comparison
    return {"unoptimized_bytes": unopt_bytes, "optimized_bytes": opt_bytes}  # Return results