23 april 2025
In deze notebook wordt stap voor stap uitgelegd hoe je in Google Colab een volledig in-memory sensorwaarschuwingssysteem kunt bouwen met FastStream, een snelle native stream-verwerkingsbibliotheek in Python. Hierbij wordt gebruikgemaakt van de RabbitMQ-integratie van FastStream, en met behulp van TestRabbitBroker is het mogelijk om de volledige verwerking lokaal te simuleren zonder externe infrastructuur. De pijplijn bestaat uit vier fasen: invoer en validatie, normalisatie, bewaking met alertering, en archivering. Voor elk van deze fases zijn datamodellen gedefinieerd met Pydantic om datakwaliteit en typeveiligheid te waarborgen.
FastStream en nest_asyncio worden geïnstalleerd om de streamverwerking en asynchrone uitvoering in Colab te ondersteunen. De nest_asyncio-module maakt het mogelijk genestelde event loops uit te voeren, terwijl het logging-framework wordt ingesteld om real-time inzicht te bieden in de pijplijnuitvoering. De stroom van berichten wordt afgehandeld via RabbitBroker, waarbij alle componenten volledig in-memory draaien voor ontwikkelings- en testdoeleinden.
Vervolgens worden drie Pydantic-datamodellen opgesteld. RawSensorData valideert de oorspronkelijke invoer, waaronder een verplicht ‘sensor_’ prefix en geldige temperatuurwaarden. NormalizedData zet de temperatuur om naar Kelvin, terwijl AlertData aangeeft of de drempel voor een waarschuwing is overschreden met een booleaanse vlag.
De datastructuur archive slaat alle waarschuwingsberichten op in een lijst. Door het gebruik van decoratoren worden de functiestappen verbonden aan specifieke broker-kanalen. Bij elke stap in de keten worden gegevens verwerkt, gelogd en doorgestuurd naar het volgende onderdeel in de pijplijn, van invoer tot en met archivering, waarin waarschuwingen met hun bijbehorende gegevens worden opgeslagen.
Een hoofdfunctie publiceert enkele voorbeeldmetingen via de TestRabbitBroker, waarna een korte pauze volgt zodat alle asynchrone functies de gegevens hebben verwerkt. De verzamelde resultaten uit de archive-lijst worden met pandas in een DataFrame geplaatst voor visuele inspectie van de waarschuwingen die zijn gegenereerd door het systeem.
Deze demonstratie laat zien hoe ontwikkelaars een schaalbare, veilige en goed testbare pipeline kunnen opzetten zonder externe afhankelijkheden. De combinatie van FastStream, Pydantic, asyncio en pandas biedt een solide basis voor real-time verwerking van gebeurtenissen, waaronder sensorbewaking en ETL-processen. Voor productietoepassingen kan eenvoudig worden overgestapt naar een echte broker en een ASGI-server worden gebruikt om de pipeline op schaal uit te rollen.
Geef een reactie