Ingest Manufacturing Sensor Streams into a Data Lakehouse with Redpanda and PyIceberg

                      
                     
"""
Production implementation for ingesting manufacturing sensor streams into a data lakehouse using Redpanda and PyIceberg.
Provides secure, scalable operations for handling real-time data.
"""

from typing import Dict, Any, List, Optional
import os
import logging
import time
import json
from kafka import KafkaProducer, KafkaConsumer
from pyiceberg import Table
from pyiceberg.exceptions import TableNotFoundError

# Set up logging configuration
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL')
    kafka_bootstrap_servers: str = os.getenv('KAFKA_BOOTSTRAP_SERVERS')
    iceberg_table_name: str = os.getenv('ICEBERG_TABLE_NAME')

# Validate input data before processing
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate incoming sensor data.
    
    Args:
        data: Input sensor data to validate
    Returns:
        bool: True if validation passes
    Raises:
        ValueError: If validation fails due to missing fields
    """
    required_fields = ['sensor_id', 'timestamp', 'value']
    for field in required_fields:
        if field not in data:
            raise ValueError(f'Missing required field: {field}')
    return True

# Function to sanitize fields in the data
async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input data fields.
    
    Args:
        data: Raw sensor data
    Returns:
        Dict: Sanitized data
    """
    return {k: str(v).strip() for k, v in data.items()}

# Function to normalize incoming sensor data
async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize sensor data for consistent processing.
    
    Args:
        data: Raw sensor data
    Returns:
        Dict: Normalized data
    """
    # Example normalization
    data['value'] = float(data['value'])  # Ensure value is float
    return data

# Function to transform records for storage
async def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform raw records into a format suitable for storage.
    
    Args:
        records: List of records to transform
    Returns:
        List: Transformed records
    """
    transformed = []
    for record in records:
        transformed.append(await normalize_data(record))
    return transformed

# Function to process a batch of sensor data
async def process_batch(batch: List[Dict[str, Any]]) -> None:
    """Process a batch of sensor data.
    
    Args:
        batch: List of sensor data records
    """
    try:
        for record in batch:
            await validate_input(record)  # Validate each record
            sanitized_record = await sanitize_fields(record)  # Sanitize fields
            # Assume we push to Kafka here
            logger.info(f'Processed record: {sanitized_record}')
    except ValueError as e:
        logger.error(f'Validation error: {e}')  # Log validation errors

# Function to fetch data from Kafka
async def fetch_data(consumer: KafkaConsumer) -> List[Dict[str, Any]]:
    """Fetch data from the Kafka topic.
    
    Args:
        consumer: Kafka consumer instance
    Returns:
        List: List of fetched records
    """
    records = []
    for message in consumer:
        records.append(json.loads(message.value))  # Deserialize JSON
    return records

# Function to save data to Iceberg
async def save_to_db(data: List[Dict[str, Any]]) -> None:
    """Save transformed data to Iceberg table.
    
    Args:
        data: List of transformed records
    """
    try:
        table = Table.load(Config.iceberg_table_name)
        table.append(data)  # Append data to the Iceberg table
        logger.info('Data saved to Iceberg successfully.')
    except TableNotFoundError:
        logger.error('Iceberg table not found.')

# Utility function to handle errors
def handle_errors(func):
    """Decorator to handle errors during function execution.
    
    Args:
        func: Function to wrap
    Returns:
        Callable: Wrapped function
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')  # Log errors
    return wrapper

# Main orchestrator class
class SensorStreamProcessor:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'sensor_data',
            bootstrap_servers=Config.kafka_bootstrap_servers,
            auto_offset_reset='earliest',
            enable_auto_commit=True,
            group_id='sensor-consumer'
        )

    async def run(self):
        while True:
            records = await fetch_data(self.consumer)  # Fetch data from Kafka
            if records:
                await process_batch(records)  # Process the fetched data
                await save_to_db(records)  # Save processed data to Iceberg
            time.sleep(1)  # Rate limit

if __name__ == '__main__':
    # Example usage
    processor = SensorStreamProcessor()
    try:
        processor.run()  # Run the processor
    except KeyboardInterrupt:
        logger.info('Shutdown requested...')  # Graceful shutdown