36 - Python for Data Lake Processing: Optimization

from typing import Iterator, Tuple  # For type annotations
import pandas as pd  # For DataFrame operations

def load_chunks(csv_path: str, chunk_size: int) -> Iterator[Tuple[pd.DataFrame, int]]:
    """
    Load CSV in chunks with type annotations.
    Args:
        csv_path: Path to CSV file.
        chunk_size: Number of rows per chunk.
    Yields:
        Tuple of DataFrame chunk and chunk index.
    """
    print(f"Loading chunks from {csv_path} with chunk_size={chunk_size}")  # Debug
    for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunk_size)):
        print(f"Processing chunk {i}: {len(chunk)} rows")  # Log chunk size
        yield chunk, i  # Yield chunk and index

# Example usage
csv_path = "data/transactions.csv"
for chunk, idx in load_chunks(csv_path, chunk_size=2):
    print(f"Chunk {idx}:")  # Debug
    print(chunk.head())  # Show chunk
    break  # Process one chunk for demo

# Expected Output:
# Loading chunks from data/transactions.csv with chunk_size=2
# Processing chunk 0: 2 rows
# Chunk 0:
#   transaction_id     product   price  quantity        date
# 0         T001  Halal Laptop  999.99         2  2023-10-01
# 1         T002   Halal Mouse   24.99        10  2023-10-02

from typing import Tuple  # For type annotations
import pandas as pd  # For DataFrame operations
import numpy as np  # For numerical operations

def validate_chunk(chunk: pd.DataFrame, min_price: float, max_quantity: int) -> Tuple[pd.DataFrame, int]:
    """
    Validate transaction chunk using vectorized operations.
    Args:
        chunk: DataFrame chunk.
        min_price: Minimum valid price.
        max_quantity: Maximum valid quantity.
    Returns:
        Tuple of validated DataFrame and invalid row count.
    """
    print(f"Validating chunk with {len(chunk)} rows")  # Debug
    # Compute amount
    chunk["amount"] = chunk["price"] * chunk["quantity"]
    # Vectorized validation
    valid_mask = (
        chunk["product"].str.startswith("Halal") &  # Halal products
        chunk["price"].ge(min_price) &  # Price >= min_price
        chunk["quantity"].le(max_quantity) &  # Quantity <= max_quantity
        chunk["price"].notna() &  # Non-null price
        chunk["quantity"].notna() &  # Non-null quantity
        np.isfinite(chunk["price"]) &  # Finite price
        np.isfinite(chunk["quantity"])  # Finite quantity
    )
    valid_chunk = chunk[valid_mask].copy()  # Filter valid rows
    invalid_count = len(chunk) - len(valid_chunk)  # Count invalid
    print(f"Valid rows: {len(valid_chunk)}, Invalid rows: {invalid_count}")  # Log
    return valid_chunk, invalid_count

# Example usage
csv_path = "data/transactions.csv"
chunk_size = 2
for chunk, _ in load_chunks(csv_path, chunk_size):
    valid_chunk, invalid_count = validate_chunk(chunk, min_price=10.0, max_quantity=100)
    print(f"Valid Chunk:\n{valid_chunk}")  # Debug
    print(f"Invalid Rows: {invalid_count}")  # Debug
    break

# Expected Output:
# Loading chunks from data/transactions.csv with chunk_size=2
# Processing chunk 0: 2 rows
# Validating chunk with 2 rows
# Valid rows: 2, Invalid rows: 0
# Valid Chunk:
#   transaction_id     product   price  quantity        date   amount
# 0         T001  Halal Laptop  999.99         2  2023-10-01  1999.98
# 1         T002   Halal Mouse   24.99        10  2023-10-02   249.90
# Invalid Rows: 0

from typing import Dict, Any  # For type annotations
import time  # For performance timing
import pandas as pd  # For DataFrame operations

def process_with_metrics(csv_path: str, chunk_size: int, min_price: float, max_quantity: int) -> Dict[str, Any]:
    """
    Process CSV with performance metrics.
    Args:
        csv_path: Path to CSV file.
        chunk_size: Number of rows per chunk.
        min_price: Minimum valid price.
        max_quantity: Maximum valid quantity.
    Returns:
        Dictionary with results and metrics.
    """
    start_time = time.time()  # Start timer
    total_valid = 0
    total_invalid = 0
    results = {"total_amount": 0.0, "chunks_processed": 0}

    for chunk, _ in load_chunks(csv_path, chunk_size):
        chunk_start = time.time()  # Start chunk timer
        valid_chunk, invalid_count = validate_chunk(chunk, min_price, max_quantity)
        total_valid += len(valid_chunk)
        total_invalid += invalid_count
        results["total_amount"] += valid_chunk["amount"].sum()
        results["chunks_processed"] += 1
        chunk_time = time.time() - chunk_start
        print(f"Chunk processed in {chunk_time:.2f} seconds")  # Log chunk time

    total_time = time.time() - start_time
    results["total_time"] = total_time
    results["total_valid"] = total_valid
    results["total_invalid"] = total_invalid
    print(f"Total processing time: {total_time:.2f} seconds")  # Log total time
    return results

# Example usage
results = process_with_metrics("data/transactions.csv", chunk_size=2, min_price=10.0, max_quantity=100)
print(f"Results: {results}")

# Expected Output (times vary):
# Loading chunks from data/transactions.csv with chunk_size=2
# Processing chunk 0: 2 rows
# Validating chunk with 2 rows
# Valid rows: 2, Invalid rows: 0
# Chunk processed in 0.01 seconds
# Processing chunk 1: 2 rows
# Validating chunk with 2 rows
# Valid rows: 1, Invalid rows: 1
# Chunk processed in 0.01 seconds
# Processing chunk 2: 1 rows
# Validating chunk with 1 rows
# Valid rows: 1, Invalid rows: 0
# Chunk processed in 0.01 seconds
# Total processing time: 0.03 seconds
# Results: {'total_amount': 2449.87, 'chunks_processed': 3, 'total_time': 0.03, 'total_valid': 4, 'total_invalid': 1}

min_price: 10.0
max_quantity: 100
required_fields:
  - transaction_id
  - product
  - price
  - quantity
  - date
product_prefix: 'Halal'
max_decimals: 2
gcs_bucket: null # Optional: GCS bucket name (e.g., "hijra-transactions")

# File: de-onboarding/utils.py (updated from Chapter 34)
from typing import Any  # For type annotations

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals.
    Handles negative numbers by removing '-' for digit check, but validation
    elsewhere ensures positive prices."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

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

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

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

def apply_valid_decimals(x: Any, 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."""
    try:
        return float(x).is_integer()
    except (ValueError, TypeError):
        return False

# File: de-onboarding/processor.py
from typing import Dict, Any, Iterator, Tuple, Optional  # For type annotations
import pandas as pd  # For DataFrame operations
import numpy as np  # For numerical operations
import yaml  # For YAML parsing
import json  # For JSON export
import time  # For performance timing
import logging  # For structured logging
import os  # For file existence check
import matplotlib.pyplot as plt  # For plotting
from google.cloud import storage  # For GCS operations

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

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

def download_from_gcs(bucket_name: Optional[str], source_blob_name: str, destination_path: str) -> bool:
    """Download file from GCS if bucket_name is provided, return True if successful."""
    if not bucket_name:
        logger.info("No GCS bucket specified, using local file")
        return False
    try:
        client = storage.Client()
        bucket = client.bucket(bucket_name)
        blob = bucket.blob(source_blob_name)
        blob.download_to_filename(destination_path)
        logger.info(f"Downloaded {source_blob_name} from GCS bucket {bucket_name} to {destination_path}")
        return True
    except Exception as e:
        logger.warning(f"Failed to download from GCS: {e}, using local file")
        return False

def load_chunks(csv_path: str, chunk_size: int) -> Iterator[Tuple[pd.DataFrame, int]]:
    """Load CSV in chunks with type annotations."""
    logger.info(f"Loading chunks from {csv_path} with chunk_size={chunk_size}")
    for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunk_size)):
        logger.info(f"Processing chunk {i}: {len(chunk)} rows")
        yield chunk, i

def validate_chunk(chunk: pd.DataFrame, config: Dict[str, Any]) -> Tuple[pd.DataFrame, int]:
    """Validate transaction chunk using vectorized operations."""
    logger.info(f"Validating chunk with {len(chunk)} rows")
    chunk["amount"] = chunk["price"] * chunk["quantity"]
    valid_mask = (
        chunk["product"].str.startswith(config["product_prefix"]) &
        chunk["price"].ge(config["min_price"]) &
        chunk["quantity"].le(config["max_quantity"]) &
        chunk["price"].notna() &
        chunk["quantity"].notna() &
        np.isfinite(chunk["price"]) &
        np.isfinite(chunk["quantity"]) &
        chunk["quantity"].apply(lambda x: isinstance(x, (int, float)) and float(x).is_integer()) &
        chunk["price"].apply(lambda x: has_valid_decimals(x, config["max_decimals"]))
    )
    valid_chunk = chunk[valid_mask].copy()
    invalid_count = len(chunk) - len(valid_chunk)
    logger.info(f"Valid rows: {len(valid_chunk)}, Invalid rows: {invalid_count}")
    return valid_chunk, invalid_count

def process_transactions(csv_path: str, config: Dict[str, Any], chunk_size: int) -> Dict[str, Any]:
    """Process transactions with performance metrics."""
    start_time = time.time()
    total_valid = 0
    total_invalid = 0
    total_amount = 0.0
    unique_products: set = set()
    chunks_processed = 0

    for chunk, _ in load_chunks(csv_path, chunk_size):
        chunk_start = time.time()
        valid_chunk, invalid_count = validate_chunk(chunk, config)
        total_valid += len(valid_chunk)
        total_invalid += invalid_count
        total_amount += valid_chunk["amount"].sum()
        unique_products.update(valid_chunk["product"].dropna().unique())
        chunks_processed += 1
        chunk_time = time.time() - chunk_start
        logger.info(f"Chunk processed in {chunk_time:.2f} seconds")

    total_time = time.time() - start_time
    results = {
        "total_amount": float(total_amount),
        "unique_products": list(unique_products),
        "total_valid": total_valid,
        "total_invalid": total_invalid,
        "chunks_processed": chunks_processed,
        "total_time_seconds": total_time
    }
    logger.info(f"Total processing time: {total_time:.2f} seconds")
    return results

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

def plot_transactions(csv_path: str, plot_path: str) -> Optional[str]:
    """Plot total transaction amounts by product."""
    df = pd.read_csv(csv_path)
    df["amount"] = df["price"] * df["quantity"]
    df = df[df["product"].str.startswith("Halal", na=False)]
    if df.empty:
        logger.info("No valid data to plot")
        return None
    # Aggregate amounts by product
    amounts = df.groupby("product")["amount"].sum()
    plt.figure(figsize=(8, 6))
    plt.bar(amounts.index, amounts)
    plt.title("Total Transaction Amounts by Product")
    plt.xlabel("Product")
    plt.ylabel("Total Amount ($)")
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    logger.info(f"Plot saved to {plot_path}")
    return plot_path

def main() -> None:
    """Main function to process transactions."""
    csv_path = "data/transactions.csv"
    config_path = "data/config.yaml"
    json_path = "data/processor_results.json"
    plot_path = "data/transactions_plot.png"
    chunk_size = 1000  # Use 1000 for scalability; smaller sizes (e.g., 2) for demonstration

    config = read_config(config_path)
    # Attempt to download from GCS if bucket is specified
    gcs_bucket = config.get("gcs_bucket")
    download_from_gcs(gcs_bucket, "transactions.csv", csv_path)

    results = process_transactions(csv_path, config, chunk_size)
    export_results(results, json_path)
    plot_transactions(csv_path, plot_path)

    logger.info("\nTransaction Report:")
    logger.info(f"Total Records Processed: {results['total_valid'] + results['total_invalid']}")
    logger.info(f"Valid Transactions: {results['total_valid']}")
    logger.info(f"Invalid Transactions: {results['total_invalid']}")
    logger.info(f"Total Amount: ${round(results['total_amount'], 2)}")
    logger.info(f"Unique Products: {results['unique_products']}")
    logger.info(f"Chunks Processed: {results['chunks_processed']}")
    logger.info(f"Processing Time: {results['total_time_seconds']:.2f} seconds")

if __name__ == "__main__":
    main()

# File: de-onboarding/tests/test_processor.py
from typing import Dict, Any  # For type annotations
import pytest  # For testing
import pandas as pd  # For DataFrame operations
import os  # For file existence check
from processor import read_config, process_transactions, export_results, download_from_gcs, plot_transactions

@pytest.fixture
def config() -> Dict[str, Any]:
    """Fixture for config."""
    return read_config("data/config.yaml")

def test_valid_input(config: Dict[str, Any], tmp_path: Any) -> None:
    """Test processing valid input."""
    results = process_transactions("data/transactions.csv", config, chunk_size=1000)
    assert results["total_amount"] == pytest.approx(2449.87, rel=1e-2)
    assert set(results["unique_products"]) == {"Halal Laptop", "Halal Mouse", "Halal Keyboard"}
    assert results["total_valid"] == 4
    assert results["total_invalid"] == 1
    assert results["chunks_processed"] == 1
    assert results["total_time_seconds"] > 0

def test_empty_input(config: Dict[str, Any], tmp_path: Any) -> None:
    """Test processing empty input."""
    empty_csv = tmp_path / "empty.csv"
    pd.DataFrame(columns=["transaction_id", "product", "price", "quantity", "date"]).to_csv(empty_csv, index=False)
    results = process_transactions(str(empty_csv), config, chunk_size=1000)
    assert results["total_amount"] == 0.0
    assert results["unique_products"] == []
    assert results["total_valid"] == 0
    assert results["total_invalid"] == 0
    assert results["chunks_processed"] == 1

def test_negative_prices(config: Dict[str, Any], tmp_path: Any) -> None:
    """Test processing input with negative prices."""
    negative_csv = tmp_path / "negative.csv"
    pd.DataFrame({
        "transaction_id": ["T001"],
        "product": ["Halal Laptop"],
        "price": [-999.99],
        "quantity": [2],
        "date": ["2023-10-01"]
    }).to_csv(negative_csv, index=False)
    results = process_transactions(str(negative_csv), config, chunk_size=1000)
    assert results["total_amount"] == 0.0
    assert results["total_valid"] == 0
    assert results["total_invalid"] == 1

def test_gcs_fallback(config: Dict[str, Any]) -> None:
    """Test GCS download fallback to local file."""
    assert download_from_gcs(None, "transactions.csv", "data/transactions.csv") == False

def test_plot_generation(config: Dict[str, Any], tmp_path: Any) -> None:
    """Test plot generation with aggregated amounts."""
    plot_path = str(tmp_path / "test_plot.png")
    result = plot_transactions("data/transactions.csv", plot_path)
    assert result == plot_path
    assert os.path.exists(plot_path)

{
  "total_amount": 2449.87,
  "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
  "total_valid": 4,
  "total_invalid": 1,
  "chunks_processed": 1,
  "total_time_seconds": 0.02
}

csv_path = "data/transactions.csv"
chunk_size = 2

csv_path = "data/transactions.csv"
chunk_size = 2
min_price = 10.0
max_quantity = 100

from typing import Dict, Any
import pandas as pd
def process_chunks(csv_path: str, chunk_size: int) -> Dict[str, Any]:
    products = []  # Bug: List causes duplicates
    for chunk, _ in load_chunks(csv_path, chunk_size):
        products.extend(chunk["product"].dropna().unique())  # Bug: extend adds duplicates
    return {"unique_products": products}
print(process_chunks("data/transactions.csv", 2))

from typing import Iterator, Tuple
import pandas as pd

def load_chunks(csv_path: str, chunk_size: int) -> Iterator[Tuple[pd.DataFrame, int]]:
    """Load CSV in chunks."""
    print(f"Loading chunks from {csv_path} with chunk_size={chunk_size}")
    for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunk_size)):
        print(f"Chunk {i}: {len(chunk)} rows")
        yield chunk, i

# Test
csv_path = "data/transactions.csv"
for chunk, idx in load_chunks(csv_path, chunk_size=2):
    print(f"Chunk {idx}:\n{chunk}")
    break

from typing import Tuple
import pandas as pd
import numpy as np

def validate_chunk(chunk: pd.DataFrame, min_price: float, max_quantity: int) -> Tuple[pd.DataFrame, int]:
    """Validate transaction chunk."""
    print(f"Validating chunk with {len(chunk)} rows")
    chunk["amount"] = chunk["price"] * chunk["quantity"]
    valid_mask = (
        chunk["product"].str.startswith("Halal") &
        chunk["price"].ge(min_price) &
        chunk["quantity"].le(max_quantity) &
        chunk["price"].notna() &
        chunk["quantity"].notna() &
        np.isfinite(chunk["price"]) &
        np.isfinite(chunk["quantity"])
    )
    valid_chunk = chunk[valid_mask].copy()
    invalid_count = len(chunk) - len(valid_chunk)
    print(f"Valid rows: {len(valid_chunk)}, Invalid rows: {invalid_count}")
    return valid_chunk, invalid_count

# Test
chunk = pd.DataFrame({
    "product": ["Halal Laptop", "Monitor"],
    "price": [999.99, -10.0],
    "quantity": [2, 5]
})
valid_chunk, invalid_count = validate_chunk(chunk, min_price=10.0, max_quantity=100)
print(f"Valid Chunk:\n{valid_chunk}")

from typing import Dict, Any
import pandas as pd
import numpy as np
import time

def process_with_metrics(csv_path: str, chunk_size: int, min_price: float, max_quantity: int) -> Dict[str, Any]:
    """Process CSV with metrics."""
    start_time = time.time()
    total_amount = 0.0
    for chunk, _ in load_chunks(csv_path, chunk_size):
        chunk_start = time.time()
        valid_chunk, _ = validate_chunk(chunk, min_price, max_quantity)
        total_amount += valid_chunk["amount"].sum()
        print(f"Chunk processed in {time.time() - chunk_start:.2f} seconds")
    total_time = time.time() - start_time
    print(f"Total time: {total_time:.2f} seconds")
    return {"total_amount": total_amount}

# Test
results = process_with_metrics("data/transactions.csv", chunk_size=2, min_price=10.0, max_quantity=100)
print(f"Total amount: {results['total_amount']}")

# Synthetic Dataset Test
def generate_synthetic_csv(path: str, rows: int = 1000) -> None:
    """Generate synthetic transactions.csv with varied data."""
    np.random.seed(42)  # For reproducibility
    products = ["Halal Laptop", "Halal Mouse", "Monitor"]
    dates = pd.date_range("2023-10-01", "2023-10-31").strftime("%Y-%m-%d").tolist()
    data = {
        "transaction_id": [f"T{i:03d}" for i in range(1, rows + 1)],
        "product": np.random.choice(products + [""], size=rows, p=[0.4, 0.4, 0.15, 0.05]),
        "price": np.random.uniform(-100, 1000, rows).round(2),
        "quantity": np.random.randint(1, 200, rows),
        "date": np.random.choice(dates, size=rows)
    }
    pd.DataFrame(data).to_csv(path, index=False)

generate_synthetic_csv("data/synthetic.csv", 1000)
results = process_with_metrics("data/synthetic.csv", chunk_size=100, min_price=10.0, max_quantity=100)
print(f"Synthetic amount: {results['total_amount']}")

from typing import Dict, Any
import pandas as pd

def process_chunks(csv_path: str, chunk_size: int) -> Dict[str, Any]:
    """Process chunks to collect unique products."""
    products = set()  # Fix