Build Manufacturing Data Pipelines with dbt and Apache Spark

                      
                     
import os
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.utils import AnalysisException

# Configuration
DB_HOST = os.getenv('DB_HOST')
DB_PORT = os.getenv('DB_PORT')
DB_NAME = os.getenv('DB_NAME')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')

# Initialize Spark session
spark = SparkSession.builder \
    .appName('Manufacturing Data Pipeline') \
    .config('spark.sql.shuffle.partitions', '200') \
    .getOrCreate()

def load_data() -> pyspark.sql.DataFrame:
    try:
        # Load data from a PostgreSQL database
        jdbc_url = f'jdbc:postgresql://{DB_HOST}:{DB_PORT}/{DB_NAME}'
        properties = {'user': DB_USER, 'password': DB_PASSWORD}
        df = spark.read.jdbc(url=jdbc_url, table='manufacturing_data', properties=properties)
        return df
    except AnalysisException as e:
        print(f'Error loading data: {e}')
        return spark.createDataFrame([], schema='')  # Return empty DataFrame on error

def transform_data(df: pyspark.sql.DataFrame) -> pyspark.sql.DataFrame:
    # Perform transformations (e.g., filtering, aggregating)
    df_transformed = df.filter(df['quantity'] > 0)  # Example filter
    return df_transformed

def save_data(df: pyspark.sql.DataFrame):
    try:
        # Save transformed data back to database
        jdbc_url = f'jdbc:postgresql://{DB_HOST}:{DB_PORT}/{DB_NAME}'
        df.write.jdbc(url=jdbc_url, table='transformed_data', mode='overwrite')
    except Exception as e:
        print(f'Error saving data: {e}')

if __name__ == '__main__':
    raw_data = load_data()
    if raw_data.count() > 0:
        transformed_data = transform_data(raw_data)
        save_data(transformed_data)