30 - Checkpoint 4: Cloud Analytics Review

from google.cloud import bigquery
from typing import Tuple

def create_dataset_and_table(project_id: str, dataset_id: str, table_id: str) -> Tuple[bigquery.Dataset, bigquery.Table]:
    """Create BigQuery dataset and fact table."""
    client = bigquery.Client(project=project_id)
    dataset_ref = client.dataset(dataset_id)
    dataset = bigquery.Dataset(dataset_ref)
    dataset.location = "US"
    dataset = client.create_dataset(dataset, exists_ok=True)
    print(f"Created dataset {dataset_id}")  # Debug

    # Define fact table schema
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
    ]
    table_ref = dataset.table(table_id)
    table = bigquery.Table(table_ref, schema=schema)

    # Example optimization (not used for small dataset)
    # table.time_partitioning = bigquery.TimePartitioning(field="date")
    # table.clustering_fields = ["product"]
    table = client.create_table(table, exists_ok=True)
    print(f"Created fact table {table_id}")  # Debug

    # Placeholder dimension table (not created, for reference; see Exercise 6 and Chapter 32 for star schema)
    # product_dim_schema = [
    #     bigquery.SchemaField("product_id", "INTEGER"),
    #     bigquery.SchemaField("product_name", "STRING"),
    # ]
    # product_dim_table = bigquery.Table(dataset.table("product_dim"), schema=product_dim_schema)

    return dataset, table

import pandas as pd
from typing import Tuple, Dict, Any, List

def load_and_validate_sales(csv_path: str, config: Dict[str, Any]) -> Tuple[pd.DataFrame, int, int, List[str]]:
    """Load and validate sales CSV, returning non-compliant products.
    Note: Uses iterrows() for simplicity; vectorized operations (Chapter 39) are more efficient for large datasets.
    Non-compliant products are printed for debugging."""
    print(f"Loading CSV: {csv_path}")
    df: pd.DataFrame = pd.read_csv(csv_path)
    print("Initial DataFrame:", df.head())

    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")
        return pd.DataFrame(), 0, len(df), []

    non_compliant_products: List[str] = []
    df = df.dropna(subset=["product"])
    for index, row in df.iterrows():
        sale = {field: str(row[field]) for field in required_fields}
        is_valid, product = validate_sale(sale, config)
        if not is_valid and product:
            non_compliant_products.append(product)

    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(lambda x: str(x).isdigit())]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(lambda x: isinstance(x, (int, float)))]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: apply_valid_decimals(x, config["max_decimals"]))]

    total_records: int = len(df)
    print("Validated DataFrame:", df)
    if non_compliant_products:
        print(f"Non-compliant products: {non_compliant_products}")
    return df, len(df), total_records, non_compliant_products

def run_analytics_query(client: bigquery.Client, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Run advanced BigQuery analytics query on fact table."""
    query = f"""
    WITH Sales AS (
        SELECT
            product,
            price * quantity AS amount
        FROM `{dataset_id}.{table_id}`
    ),
    RankedSales AS (
        SELECT
            product,
            amount,
            ROW_NUMBER() OVER (PARTITION BY product ORDER BY amount DESC) AS rank
        FROM Sales
    )
    SELECT
        product,
        amount AS top_sale
    FROM RankedSales
    WHERE rank = 1
    """
    query_job = client.query(query)
    results = query_job.to_dataframe()

    if results.empty:
        print("No data returned from query")  # Handle edge case
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}

    total_sales = np.sum(results["top_sale"].values)
    unique_products = results["product"].unique().tolist()
    top_products = results.set_index("product")["top_sale"].to_dict()

    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }

def test_load_sales():
    """Test sales loading and validation."""
    config = {"product_prefix": "Halal", "max_quantity": 100, "min_price": 10.0}
    df, valid, total, non_compliant = load_and_validate_sales("data/sample.csv", config)
    assert len(df) == 2
    assert valid == 2
    assert total == 2
    assert not non_compliant  # No non-compliant products in sample.csv

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

# File: de-onboarding/utils.py
from typing import Union, Dict, Any, List, Tuple

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

def clean_string(s: str) -> str:
    """Strip whitespace from string."""
    return s.strip()

def is_numeric_value(x: Any) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))

def has_valid_decimals(x: Union[int, float], max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)

def apply_valid_decimals(x: Union[int, float], max_decimals: int) -> bool:
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: Any) -> bool:
    """Check if value is an integer when converted to string."""
    return str(x).isdigit()

def validate_sale(sale: Dict[str, str], config: Dict[str, Any]) -> Tuple[bool, str]:
    """Validate sale based on config rules, returning compliance status and product name if non-compliant.
    Inputs are strings from CSV parsing; Pandas type inference is handled in load_and_validate_sales."""
    required_fields = config["required_fields"]
    min_price = config["min_price"]
    max_quantity = config["max_quantity"]
    prefix = config["product_prefix"]
    max_decimals = config["max_decimals"]

    print(f"Validating sale: {sale}")
    product = clean_string(sale["product"])

    for field in required_fields:
        if not sale[field] or sale[field].strip() == "":
            print(f"Invalid sale: missing {field}: {sale}")
            return False, product

    if not product.startswith(prefix):
        print(f"Invalid sale: product lacks '{prefix}' prefix: {sale}")
        return False, product

    price = clean_string(sale["price"])
    if not is_numeric(price, max_decimals) or float(price) < min_price or float(price) <= 0:
        print(f"Invalid sale: invalid price: {sale}")
        return False, product

    quantity = clean_string(sale["quantity"])
    if not quantity.isdigit() or int(quantity) > max_quantity:
        print(f"Invalid sale: invalid quantity: {sale}")
        return False, product

    return True, ""

# File: de-onboarding/analytics_tool.py
import pandas as pd
import numpy as np
import yaml
import json
import matplotlib.pyplot as plt
from google.cloud import bigquery
import os
from typing import Dict, Tuple, Any, List

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    print(f"Opening config: {config_path}")
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    print(f"Loaded config: {config}")
    return config

def load_and_validate_sales(csv_path: str, config: Dict[str, Any]) -> Tuple[pd.DataFrame, int, int, List[str]]:
    """Load and validate sales CSV, returning non-compliant products.
    Note: Uses iterrows() for simplicity; vectorized operations (Chapter 39) are more efficient for large datasets.
    Non-compliant products are printed for debugging."""
    print(f"Loading CSV: {csv_path}")
    df: pd.DataFrame = pd.read_csv(csv_path)
    print("Initial DataFrame:", df.head())

    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")
        return pd.DataFrame(), 0, len(df), []

    non_compliant_products: List[str] = []
    df = df.dropna(subset=["product"])
    for index, row in df.iterrows():
        sale = {field: str(row[field]) for field in required_fields}
        is_valid, product = validate_sale(sale, config)
        if not is_valid and product:
            non_compliant_products.append(product)

    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(lambda x: str(x).isdigit())]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(lambda x: isinstance(x, (int, float)))]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: apply_valid_decimals(x, config["max_decimals"]))]

    total_records: int = len(df)
    print("Validated DataFrame:", df)
    if non_compliant_products:
        print(f"Non-compliant products: {non_compliant_products}")
    return df, len(df), total_records, non_compliant_products

def load_to_bigquery(df: pd.DataFrame, project_id: str, dataset_id: str, table_id: str) -> None:
    """Load DataFrame to BigQuery fact table."""
    client = bigquery.Client(project=project_id)
    dataset, table = create_dataset_and_table(project_id, dataset_id, table_id)
    table_ref = f"{project_id}.{dataset_id}.{table_id}"

    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_TRUNCATE",
        schema=[
            bigquery.SchemaField("product", "STRING"),
            bigquery.SchemaField("price", "FLOAT"),
            bigquery.SchemaField("quantity", "INTEGER"),
        ],
    )
    job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
    job.result()
    print(f"Loaded {len(df)} rows to {table_ref}")

def create_dataset_and_table(project_id: str, dataset_id: str, table_id: str) -> Tuple[bigquery.Dataset, bigquery.Table]:
    """Create BigQuery dataset and fact table."""
    client = bigquery.Client(project=project_id)
    dataset_ref = client.dataset(dataset_id)
    dataset = bigquery.Dataset(dataset_ref)
    dataset.location = "US"
    dataset = client.create_dataset(dataset, exists_ok=True)
    print(f"Created dataset {dataset_id}")

    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
    ]
    table_ref = dataset.table(table_id)
    table = bigquery.Table(table_ref, schema=schema)

    # Example optimization (not used for small dataset)
    # table.time_partitioning = bigquery.TimePartitioning(field="date")
    # table.clustering_fields = ["product"]
    table = client.create_table(table, exists_ok=True)
    print(f"Created fact table {table_id}")

    # Placeholder dimension table (not created, for reference; see Exercise 6 and Chapter 32 for star schema)
    # product_dim_schema = [
    #     bigquery.SchemaField("product_id", "INTEGER"),
    #     bigquery.SchemaField("product_name", "STRING"),
    # ]
    # product_dim_table = bigquery.Table(dataset.table("product_dim"), schema=product_dim_schema)

    return dataset, table

def run_analytics_query(client: bigquery.Client, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Run advanced analytics query on fact table."""
    query = f"""
    WITH Sales AS (
        SELECT
            product,
            price * quantity AS amount
        FROM `{dataset_id}.{table_id}`
    ),
    RankedSales AS (
        SELECT
            product,
            amount,
            ROW_NUMBER() OVER (PARTITION BY product ORDER BY amount DESC) AS rank
        FROM Sales
    )
    SELECT
        product,
        amount AS top_sale
    FROM RankedSales
    WHERE rank = 1
    """
    query_job = client.query(query)
    results = query_job.to_dataframe()

    if results.empty:
        print("No data returned from query")  # Handle edge case
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}

    total_sales = np.sum(results["top_sale"].values)
    unique_products = results["product"].unique().tolist()
    top_products = results.set_index("product")["top_sale"].to_dict()

    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export results to JSON."""
    print(f"Writing to: {json_path}")
    with open(json_path, "w") as file:
        json.dump(results, file, indent=2)
    print(f"Exported results to {json_path}")

def plot_sales(df: pd.DataFrame, plot_path: str) -> None:
    """Generate sales trend plot."""
    if df.empty:
        print("No data to plot")
        return

    df["amount"] = df["price"] * df["quantity"]
    plt.figure(figsize=(8, 6))
    plt.bar(df["product"], df["amount"])
    plt.title("Sales by Product")
    plt.xlabel("Product")
    plt.ylabel("Sales Amount ($)")
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to {plot_path}")
    print(f"File exists: {os.path.exists(plot_path)}")

def main() -> None:
    """Main function to process sales data."""
    csv_path = "data/sales.csv"
    config_path = "data/config.yaml"
    json_path = "data/analytics_results.json"
    plot_path = "data/sales_trend.png"
    project_id = "your-project-id"  # Replace with your Google Cloud project ID
    dataset_id = "sales_dataset"
    table_id = "sales_fact"  # Fact table

    config = read_config(config_path)
    df, valid_sales, total_records, non_compliant = load_and_validate_sales(csv_path, config)
    if not df.empty:
        load_to_bigquery(df, project_id, dataset_id, table_id)
        client = bigquery.Client(project=project_id)
        results = run_analytics_query(client, dataset_id, table_id)
        export_results(results, json_path)
        plot_sales(df, plot_path)

    print("\nSales Analytics Report:")
    print(f"Total Records Processed: {total_records}")
    print(f"Valid Sales: {valid_sales}")
    print(f"Invalid Sales: {total_records - valid_sales}")
    print(f"Total Sales: ${round(results['total_sales'], 2)}")
    print(f"Unique Products: {results['unique_products']}")
    print(f"Top Products: {results['top_products']}")
    print("Processing completed")

if __name__ == "__main__":
    main()

{
  "total_sales": 2499.83,
  "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
  "top_products": {
    "Halal Laptop": 1999.98,
    "Halal Mouse": 249.9,
    "Halal Keyboard": 249.95
  }
}

def test_load_sales():
    config = read_config("data/config.yaml")
    df, valid, total, non_compliant = load_and_validate_sales("data/sample.csv", config)
    assert len(df) == 3  # Bug: Wrong count

SELECT SUM(price * quantity) AS total_sales
FROM `sales_dataset.sales_fact`;

from google.cloud import bigquery
from typing import Tuple

def create_dataset_and_table(project_id: str, dataset_id: str, table_id: str) -> Tuple[bigquery.Dataset, bigquery.Table]:
    """Create BigQuery dataset and fact table."""
    client = bigquery.Client(project=project_id)
    dataset_ref = client.dataset(dataset_id)
    dataset = bigquery.Dataset(dataset_ref)
    dataset.location = "US"
    dataset = client.create_dataset(dataset, exists_ok=True)
    print(f"Created dataset {dataset_id}")

    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER"),
    ]
    table_ref = dataset.table(table_id)
    table = bigquery.Table(table_ref, schema=schema)
    table = client.create_table(table, exists_ok=True)
    print(f"Created fact table {table_id}")
    return dataset, table

# Test
create_dataset_and_table("your-project-id", "test_dataset", "test_sales_fact")

import pandas as pd
from typing import Tuple, List

def load_and_validate_sales(csv_path: str, config: dict) -> Tuple[pd.DataFrame, int, int, List[str]]:
    """Load and validate sales CSV."""
    df: pd.DataFrame = pd.read_csv(csv_path)
    df = df[df["product"].str.startswith(config["product_prefix"])]
    total_records: int = len(df)
    return df, len(df), total_records, []

# Test
config = {"product_prefix": "Halal"}
df, valid, total, non_compliant = load_and_validate_sales("data/sample.csv", config)
print(f"DataFrame with {len(df)} rows")

from google.cloud import bigquery
from typing import Dict

def run_analytics_query(client: bigquery.Client, dataset_id: str, table_id: str) -> Dict[str, Any]:
    """Run advanced analytics query on fact table."""
    query = f"""
    SELECT
        product,
        price * quantity AS amount
    FROM `{dataset_id}.{table_id}`
    """
    query_job = client.query(query)
    results = query_job.to_dataframe()
    return {"products": results["product"].tolist(), "amounts": results["amount"].tolist()}

# Test
client = bigquery.Client(project="your-project-id")
results = run_analytics_query(client, "sales_dataset", "sales_fact")
print("DataFrame with top sales per product")

import pandas as pd
import matplotlib.pyplot as plt

def plot_sales(df: pd.DataFrame, plot_path: str) -> None:
    """Generate sales plot."""
    if df.empty:
        print("No data to plot")
        return
    df["amount"] = df["price"] * df["quantity"]
    plt.figure(figsize=(8, 6))
    plt.bar(df["product"], df["amount"])
    plt.title("Sales by Product")
    plt.xlabel("Product")
    plt.ylabel("Sales Amount ($)")
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to {plot_path}")

# Test
df = pd.read_csv("data/sample.csv")
plot_sales(df, "data/plot.png")

import pytest
from analytics_tool import load_and_validate_sales, read_config

def test_load_sales():
    """Test sales loading and validation."""
    config = read_config("data/config.yaml")
    df, valid, total, non_compliant = load_and_validate_sales("data/sample.csv", config)
    assert len(df) == 2  # Fix: Correct count
    assert valid == 2
    assert total == 2
    assert not non_compliant

# Run: pytest ex5_test.py

def create_star_schema_description(output_path: str) -> None:
    """Generate star schema description and Mermaid diagram for sales.csv, debugging incorrect column types."""
    # Debug: Fix incorrect diagram (e.g., product: INTEGER -> product: STRING)
    schema = """
Star Schema for sales.csv:
- Fact Table: sales_fact
  - Columns: product (STRING), price (FLOAT), quantity (INTEGER)
- Dimension Table: product_dim
  - Columns: product_id (INTEGER), product_name (STRING)

Mermaid Diagram:
erDiagram
    sales_fact ||--o{ product_dim : references
    sales_fact {
        string product
        float price
        integer quantity
    }
    product_dim {
        integer product_id
        string product_name
    }
"""
    with open(output_path, "w") as file:
        file.write(schema)
    print(f"Schema saved to {output_path}")

# Test
create_star_schema_description("ex6_schema.txt")

def write_bigquery_comparison(concepts_path: str, query_path: str) -> None:
    """Write comparison of BigQuery vs. SQLite and total sales query."""
    comparison = """
BigQuery is ideal for Hijra Group’s large-scale sales analytics due to its columnar storage, which optimizes analytical queries by storing data by column, enabling faster aggregations compared to SQLite’s row-based storage. BigQuery’s parallel query execution across distributed nodes handles millions of sales records efficiently, unlike SQLite, which is limited to single-threaded processing on a local machine. This scalability ensures Hijra Group can validate Halal products across vast transaction datasets quickly, supporting real-time analytics for Sharia-compliant fintech solutions.
"""
    query = """
SELECT SUM(price * quantity) AS total_sales
FROM `sales_dataset.sales_fact`;
"""
    with open(concepts_path, "w") as file:
        file.write(comparison)
    with open(query_path, "w") as file:
        file.write(query)
    print(f"Comparison saved to {concepts_path}")
    print(f"Query saved to {query_path}")

# Test
write_bigquery_comparison("ex7_concepts.txt", "ex7_query.sql")