Build Manufacturing Data Pipelines with dbt and Apache Spark

                      
                     
"""
Production implementation for building manufacturing data pipelines using dbt and Apache Spark.
Provides secure, scalable operations for ETL processes.
"""

from typing import Dict, Any, List
import os
import logging
import time
from pyspark.sql import SparkSession
from pyspark.sql import DataFrame

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

class Config:
    """
    Configuration class for environment variables.
    """
    db_url: str = os.getenv('DATABASE_URL')
    db_user: str = os.getenv('DATABASE_USER')
    db_password: str = os.getenv('DATABASE_PASSWORD')

class SparkSessionManager:
    """
    Context manager for Spark session.
    """
    def __enter__(self) -> SparkSession:
        spark = SparkSession.builder \
            .appName("ManufacturingDataPipeline") \
            .getOrCreate()
        return spark

    def __exit__(self, exc_type, exc_value, traceback):
        logger.info("Stopping Spark session.")
        SparkSession.stop()

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for the pipeline.
    
    Args:
        data: Input data to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'manufacturing_id' not in data:
        raise ValueError('Missing manufacturing_id')
    return True

async def fetch_data(spark: SparkSession, query: str) -> DataFrame:
    """Fetch data from the database using Spark SQL.
    
    Args:
        spark: Active Spark session
        query: SQL query to execute
    Returns:
        DataFrame containing the result set
    """
    logger.info(f"Fetching data with query: {query}")
    return spark.sql(query)

async def transform_records(df: DataFrame) -> DataFrame:
    """Transform raw records into a structured format.
    
    Args:
        df: Raw DataFrame to transform
    Returns:
        Transformed DataFrame
    """
    logger.info(f"Transforming records.")
    return df.select("manufacturing_id", "value", "timestamp")

async def aggregate_metrics(df: DataFrame) -> DataFrame:
    """Aggregate metrics for analysis.
    
    Args:
        df: DataFrame containing transformed records
    Returns:
        Aggregated DataFrame
    """
    logger.info(f"Aggregating metrics.")
    return df.groupBy("manufacturing_id").agg({"value": "avg"})

async def save_to_db(df: DataFrame, table_name: str) -> None:
    """Save DataFrame to the database.
    
    Args:
        df: DataFrame to save
        table_name: Name of the target table
    Raises:
        Exception: If save operation fails
    """
    logger.info(f"Saving DataFrame to {table_name}.")
    try:
        df.write \
            .format("jdbc") \
            .option("url", Config.db_url) \
            .option("dbtable", table_name) \
            .option("user", Config.db_user) \
            .option("password", Config.db_password) \
            .save()
    except Exception as e:
        logger.error(f"Failed to save DataFrame: {e}")
        raise

async def run_pipeline(query: str) -> None:
    """Main orchestration function for the data pipeline.
    
    Args:
        query: SQL query for fetching data
    """
    try:
        async with SparkSessionManager() as spark:
            # Fetch data from the source
            raw_data = await fetch_data(spark, query)
            # Validate the input data
            await validate_input(raw_data)
            # Transform records for analysis
            transformed_data = await transform_records(raw_data)
            # Aggregate metrics
            aggregated_data = await aggregate_metrics(transformed_data)
            # Save aggregated data to the database
            await save_to_db(aggregated_data, "aggregated_metrics")
    except Exception as error:
        logger.error(f"Pipeline run failed: {error}")

if __name__ == '__main__':
    # Sample SQL query
    sql_query = "SELECT * FROM manufacturing_data WHERE timestamp > NOW() - INTERVAL 1 DAY"
    # Run the pipeline
    run_pipeline(sql_query)