Stream Industrial Sensor Telemetry to Iceberg Tables with Apache Kafka and pyiceberg

"""
Production implementation for streaming industrial sensor telemetry to Iceberg tables.
Provides secure, scalable operations utilizing Apache Kafka and pyiceberg.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import json
import time
from kafka import KafkaProducer, KafkaConsumer
from pyiceberg import Table

# Setup logging for monitoring the process.
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    KAFKA_TOPIC: str = os.getenv('KAFKA_TOPIC', 'sensor_data')
    KAFKA_BOOTSTRAP_SERVERS: str = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
    ICEBERG_TABLE: str = os.getenv('ICEBERG_TABLE', 'default.sensor_table')

class SensorTelemetry:
    def __init__(self) -> None:
        self.producer = KafkaProducer(bootstrap_servers=Config.KAFKA_BOOTSTRAP_SERVERS)
        self.consumer = KafkaConsumer(
            Config.KAFKA_TOPIC,
            bootstrap_servers=Config.KAFKA_BOOTSTRAP_SERVERS,
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )
        self.table = Table.from_uri(Config.ICEBERG_TABLE)

    def validate_input(self, data: Dict[str, Any]) -> bool:
        """Validate sensor data input.
        
        Args:
            data: Input data to validate
        Returns:
            True if valid
        Raises:
            ValueError: If validation fails
        """
        if 'sensor_id' not in data:
            raise ValueError('Missing sensor_id')
        if 'timestamp' not in data:
            raise ValueError('Missing timestamp')
        if 'value' not in data:
            raise ValueError('Missing value')
        return True

    def transform_records(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Transform incoming sensor data into a format suitable for Iceberg.
        
        Args:
            data: Raw sensor data
        Returns:
            Transformed data ready for processing
        """
        return {
            'sensor_id': data['sensor_id'],
            'timestamp': data['timestamp'],
            'value': float(data['value']),
        }

    def process_batch(self, records: List[Dict[str, Any]]) -> None:
        """Process a batch of sensor records and save to Iceberg table.
        
        Args:
            records: A list of transformed sensor records
        """
        for record in records:
            self.table.append(record)  # Append each record to the Iceberg table

    def fetch_data(self) -> None:
        """Fetch data from Kafka topic and process it.
        """
        logger.info('Starting to consume messages from Kafka...')
        for message in self.consumer:
            try:
                raw_data = message.value
                if self.validate_input(raw_data):  # Validate data
                    transformed_data = self.transform_records(raw_data)  # Transform data
                    self.process_batch([transformed_data])  # Process and save data
                    logger.info(f'Processed record: {transformed_data}')
            except Exception as e:
                logger.error(f'Error processing message: {e}')

    def send_data(self, data: Dict[str, Any]) -> None:
        """Send sensor data to Kafka topic.
        
        Args:
            data: Sensor data to send
        """
        try:
            self.producer.send(Config.KAFKA_TOPIC, value=data)
            self.producer.flush()  # Ensure the message is sent
            logger.info(f'Sent data: {data}')
        except Exception as e:
            logger.error(f'Error sending data to Kafka: {e}')

if __name__ == '__main__':
    # Example usage of the SensorTelemetry class
    telemetry = SensorTelemetry()
    # Simulating sending sensor data
    sample_data = {'sensor_id': '1234', 'timestamp': '2023-10-01T12:00:00Z', 'value': 42.0}
    telemetry.send_data(sample_data)
    telemetry.fetch_data()  # Start consuming data from Kafka