Stream IoT Sensor Data into Lakehouse Tables with Kafka and Flink CDC

                      
                     
"""
Production implementation for Streaming IoT Sensor Data into Lakehouse Tables with Kafka and Flink CDC.
Provides secure, scalable operations.
"""

from typing import Dict, Any, List
import os
import logging
import time
from kafka import KafkaConsumer, KafkaProducer
import json
import psycopg2
from contextlib import contextmanager

# Logger setup for tracking application behavior
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

@contextmanager
def db_connection() -> Any:
    """
    Context manager for database connection.
    
    Yields:
        Connection object
    """
    conn = psycopg2.connect(
        dbname=Config.db_url,
        user=Config.db_user,
        password=Config.db_password
    )
    try:
        yield conn
    finally:
        conn.close()  # Close the connection safely

def validate_input(data: Dict[str, Any]) -> bool:
    """
    Validate incoming data from IoT sensors.
    
    Args:
        data: Input data dictionary
    Returns:
        bool: True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'sensor_id' not in data:
        raise ValueError('Missing sensor_id')
    if 'value' not in data:
        raise ValueError('Missing value')
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sanitize fields in the incoming data.
    
    Args:
        data: Input data dictionary
    Returns:
        Dict[str, Any]: Sanitized data
    """
    return {key: str(value).strip() for key, value in data.items()}

def transform_records(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Transform data into the format required for Lakehouse.
    
    Args:
        data: Input sensor data
    Returns:
        Dict[str, Any]: Transformed data
    """
    return {
        'sensor_id': data['sensor_id'],
        'value': float(data['value']),
        'timestamp': time.time()
    }

def fetch_data() -> List[Dict[str, Any]]:
    """
    Fetch data from Kafka stream.
    
    Returns:
        List[Dict[str, Any]]: List of sensor data
    """
    consumer = KafkaConsumer(
        'sensor_data',
        bootstrap_servers=Config.kafka_broker,
        auto_offset_reset='earliest',
        group_id='sensor-consumer'
    )
    data = []
    for message in consumer:
        data.append(json.loads(message.value))
        logger.info(f"Fetched message: {message.value}")
    return data

def save_to_db(data: Dict[str, Any]) -> None:
    """
    Save transformed data into the Lakehouse.
    
    Args:
        data: Transformed sensor data
    """
    with db_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'INSERT INTO lakehouse_table (sensor_id, value, timestamp) VALUES (%s, %s, %s)',
                (data['sensor_id'], data['value'], data['timestamp'])
            )
            conn.commit()
            logger.info(f"Saved data: {data}")

def process_batch(data_batch: List[Dict[str, Any]]) -> None:
    """
    Process a batch of sensor data.
    
    Args:
        data_batch: List of sensor data
    """
    for data in data_batch:
        try:
            validate_input(data)  # Validate incoming data
            sanitized = sanitize_fields(data)  # Sanitize fields
            transformed = transform_records(sanitized)  # Transform data
            save_to_db(transformed)  # Save to the database
        except ValueError as e:
            logger.error(f"Validation error: {e}")  # Log validation errors
        except Exception as e:
            logger.error(f"Error processing data: {e}")  # Log other errors

def main() -> None:
    """
    Main function to orchestrate data streaming and processing.
    """
    while True:
        data_batch = fetch_data()  # Fetch data from Kafka
        process_batch(data_batch)  # Process the fetched data
        time.sleep(1)  # Wait before the next batch

if __name__ == '__main__':
    # Run the main function
    main()