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

                      
                      
import os
import asyncio
from aiokafka import AIOKafkaConsumer
from pyiceberg import Table

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

# Initialize Iceberg table
iceberg_table = Table.for_name(ICEBERG_TABLE)

# Consumer setup
async def consume_sensor_data() -> None:
    consumer = AIOKafkaConsumer(
        'sensors',
        bootstrap_servers=KAFKA_BROKER,
        group_id='sensor_group',
        auto_offset_reset='earliest',
    )
    await consumer.start()
    try:
        async for msg in consumer:
            await process_message(msg.value)
    except Exception as e:
        print(f'Error: {e}')
    finally:
        await consumer.stop()

# Process incoming messages
async def process_message(message: bytes) -> None:
    try:
        # Assume message is in JSON format
        sensor_data = json.loads(message.decode('utf-8'))
        # Save to Iceberg table
        iceberg_table.append(sensor_data)
    except Exception as e:
        print(f'Failed to process message: {e}')

if __name__ == '__main__':
    asyncio.run(consume_sensor_data())