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

                      
                      from typing import Dict, Any
import os
import time
import json
from redpanda import Producer
from pyiceberg import IcebergTable

# Configuration
KAFKA_BROKER = os.getenv('KAFKA_BROKER', 'localhost:9092')
ICEBERG_TABLE = os.getenv('ICEBERG_TABLE', 'manufacturing_sensors')

# Initialize Redpanda Producer
producer = Producer({'bootstrap_servers': KAFKA_BROKER})

# Iceberg Table Initialization
iceberg_table = IcebergTable(ICEBERG_TABLE)

def ingest_sensor_data(sensor_id: str, data: Dict[str, Any]) -> None:
    try:
        # Prepare data
        sensor_data = {
            'sensor_id': sensor_id,
            'timestamp': time.time(),
            'data': data
        }
        # Publish to Redpanda
        producer.send('sensor_streams', json.dumps(sensor_data).encode('utf-8'))
        print(f'Sent data for sensor {sensor_id}: {sensor_data}')
        # Write to Iceberg table
        iceberg_table.write(sensor_data)
    except Exception as error:
        print(f'Error ingesting data: {str(error)}')

if __name__ == '__main__':
    # Example sensor data ingestion
    sample_data = {'temperature': 22.5, 'pressure': 101.3}
    ingest_sensor_data('sensor_1', sample_data)