SBTB 2023: Danica Fine, Practical Pipelines: A Houseplant Alerting System with ksqlDB.

SBTB 2023: Danica Fine, Practical Pipelines: A Houseplant Alerting System with ksqlDB.

Recording: SBTB 2023: Danica Fine, Practical Pipelines: A Houseplant Alerting System with ksqlDB.

[Music] excited to be here today I I'm I'm local actually and I've been monitoring this conference for years I'm excited to finally be here uh to share what I know about Kafka and hopefully get you excited about this technology so I'm going to wake you up I know it's the end of the day um but I am very loud and I'm very enthusiastic about house plants and Kafka um so yeah I work for Confluence we do a lot of Kafka we do a lot of real-time data streaming and so I I just love to talk about those things but I'm a well-rounded human I do things outside of tech and the only other thing I do is house plants um case in point here are some of my house plants that I own I have about 40 or 50 of them strewn about my house um how many of you have house plants I feel like yeah most of you you're here this late in the day you came for the house plants not for the Kafka I get it that's fine um so this is going to resonate with you right house plants sometimes we just make it look easy you know but sometimes they're difficult um I got a lot of my house plants over the pandemic and I had 24/7 to take care of them and that was wonderful you know you could check on them 45 times a day and and do what you needed to do with them but I live my life now sometimes I go to Oakland uh instead of hanging out in the peninsula and so occasionally I have house plants that turn out like this they do not they're not supposed to look like that right that's a sad sad house plant and so I want to keep living my life I want to live my best life here and so I asked myself if there's a better way right I need to take care of my house plants I need a better way to monitor them that doesn't involve me going around and checking on them multiple times per day right and I know what you're thinking of course there's a better way Amazon sells 45,000 different ways to monitor your house plants and they can be delivered same day right um but that wouldn't make a good story you're here to hear about Kafka and something more interesting than that right many of you are Engineers um you are a curious person maybe even a creative person you showed up at this talk and so that's the wrong question this is probably more accurate right um before I became a developer Advocate I spent a number of years building out streaming data pipelines with CFA streams and so when I became a developer Advocate I promised myself that I was going to build interesting things things that solve problems that I had uh because then it's more fun to talk about it and it's a win-win I get something cool out of it too um and so this was also an opportunity for me to build out an iot project get a Raspberry Pi and play around with it so this is the perfect storm I could solve a problem that I had with my house plants I could use a Pachi kafa at home who does that we're going to do it um and then I could also build out a project with a Raspberry Pi this is great so let's get into it stock design what does this look like so I figured there's a lot of ways that you can monitor house plants a lot of different metrics that you can look at determine how you need to care for them but the biggest one for me is okay what's their moisture level right do they need to be watered right now I don't want droopy sad plants okay and so I would get some soil moisture sensors put those in those plants connect that to the raspberry pie capture those readings throw it into kovka we'll deal with it later throw it into cavka that's probably not good enough though we need a little more information to actually make sense of those readings so I'm going to call that metadata for now we got to get that and throw that into Kafka too both those data sets we can then do some stream processing with those combine those and at the end I am a lazy person I'm not going to say lazy um I like convenience and so I want to receive an alert at the end on my phone telling me when I need to take care of my plans great so before we get into that though how many of you have worked with Kafka before okay some of you didn't raise your hand and we can talk later too um but for those of you who did not raise your hand um just want to level set on what this is as a high level so we all understand so Kafka is a distributed event streaming platform if I had more time I would go into the nitty-gritty details of this definition more um but basically what you need to know is kfka is a Swiss army knife of sorts it's a unified high throughput low latency platform for handling your real-time data and allowing you then to build reactive applications that can then tap into that information in real time all right it's pretty cool so we're going to use this as the backbone of this system to store this data and then allow us to do some stream processing on top of that real-time information okay so going back to me in my cool project um so what do we need to do to actually make this come together well first things first we need some Kafka all right so Kafka is a distributed system we need some sort of cluster with multiple nodes sounds like an infrastructure nightmare I don't like managing Kafka and here's my Shameless plug I work for confluent and we offer Kafka fully managed in the cloud um beyond that um there's some cool things that we have built on top of that tools for integrating with external data sources in syncs with caka connect stream processing built into our platform with kle DB and um also schema management so we're going to get into all of these eventually um I'm going to use this just because I'm going to spend a lot of time in this project managing some actual Hardware I don't also want to manage some infrastructure so if you want to follow along at home as an exercise to the reader if you want to build this yourself just know that you have to have a Kafka cluster running somewhere I don't care where you get it cool so we've got a cocka cluster did some hand waving there let's go and dive into the physical system this is I had the most fun with this it was my first Hardware project and I decided to cut and crimp all of my own wires it was really really fun um this is not a hardware talk though this is not a hardware conference but if you do want to follow along and build something here are you know the high level the things that you would need to make this system yourself um I do want to go into the soil moisture sensors that I chose though because they're going to come up later they caused me a lot of headaches so these are iqu C capacitive moisture sensors um they're about a dollar each so pretty pretty cheap I could have spent more money on them maybe I would have gotten a better sensor but the cool thing about these so they're I squared C if you don't work with any sort of iot stuff or Hardware components um I squ seed components communicate using a unique address which is set at the component level at the sensor level and so all these sensors are wired together to the breadboard and there's a single set of wires that communicate from the breadboard into the Raspberry Pi and so that unique address is really important so you know which sensor you're pulling that information from right unfortunately the sensors that I chose had a fun limitation that I found out after buying like 15 of them um in that the I squared C address could only be set to one of four values cool so for my 40 or 50 plants I can only hook up up four plants at a time with the system um yes there are ways around this it's also beyond the scope of me and my skills uh for my first Hardware project um but this is going to come up we are going to come up with a a way to get around this later on in the system um but just keep that in mind it's a fun thing to consider uh if you're building something for yourself so here's again a little bit more hand waving I built the system hooked up the sensors um and I'm ready to collect data from these plants all I had to do now was get that data into kfka that's it just just put the data in cka but before we get ahead of ourselves anytime I build any sort of project I do I'm kind of a nerd I like to sit down and actually write a schema um it's a best practice that I just have drilled into my head um and it's a good best practice to adhere to okay uh it's going to save a lot of headaches you can actually reason about the information that you're working with and make better decisions hopefully so I defined an AOS schema for these readings that I'm collecting from the moisture sensor ansers pretty simple um schema we have the percentage moisture that's really important I also got temperature for free with those uh Those sensors so I tossed that in there as well and then I added a plant ID field that I was going to just manually assign to these plants and I'm going to use that as the key of this information as we put it into the Kafka topics all right so this is how the data should look but how do we get it in there well there's a couple different ways to get information into Kafka either the producer API or Kafka connect producer API this is your low-level sort of vanilla way of getting information into Kafka it's great because it's available in all of your favorite languages and some of your least favorite so you know you have options you'll want to use the producer API if you either own or have access to the system or the application that's generating your data um so if I wanted to capture you know specific log events on the back end of some website yeah I'll just toss in uh a producer there and just throw it into kovka that'd be pretty easy the other option is Kafka connect as the name implies Kafka connect is a way to connect Kafka with external sources and also syncs later on um what's really great about it is that it's no code it's configuration driven um there are dozens hundreds of pre-made connectors out there for different sources and syncs that you can utilize and so you write up a a config for it point to the data source or sync how to authenticate with it if it's a database which tables or what query you want to execute against that datab space and when you start running that connector it's going to start pulling that information into kfka identifying it if you will I'm still working on that term not sure if anyone wants to use it but it's there um and just making that data at rest a little more real time right so what did I use in my case well I own the Raspberry Pi I am my only stakeholder and I'm writing the script so we're going to use the producer API here so I'm going to use the conflent kfka python library for my producers um you'll recall that tiny little issue with the uh the four sensors that I could have hooked up at a time um I need to reference each of these moisture sensors sensors by their address so I'm going to hardcode Red Flag I'm going to hardcode a mapping of the sensor address to the plant ID that that sensor is currently in all right we're going to come back to that because that's I should not be doing that um and then within the script we are looping every 30 seconds over that mapping of address imp plant ID we are fetching that information the moisture Computing the percent moisture grabbing the temperature as well and packaging it according to that schema we defined and we're just going to produce it into kavka cool pretty easy so those readings are in kavka that's wonderful that's that's half the battle um but that information isn't good enough on its own right now I have a bunch of data that just say okay what are these moisture readings but every plant is different every plant has different moisture requirements and so I need a little bit more information to actually accurately make sense of this these readings okay so we're g to use another data set some metadata okay and so I created another AOS schema again this one's a little more interesting got a couple more fields in here because I couldn't stop myself um we've got a lot of the scientific name the common name of the plant the given name of the plant because all of my plants have names um the low and high moisture threshold of that plant is comfortable within with uh within the uh same for the temperature values as well and of course had to include the plant ID in there so that we can join it onto the other data set later on so we have that schema we know how it should look what should we use to get into kavka well producer API kavka connect which one so kavka connect it's really really great for slow changing data data at rest that is in a database or at the other end of an API call um metadata or reference data sets like we're dealing with here that's probably something that should live in a database base right but I only have four plants that I need to deal with right now and at most I'm only going to have 40 or 50 rows in this database so I don't know it seems like Overkill from an infrastructure standpoint for me to waste time putting this data uh in maintaining another database so let's store the data in Kafka that is a completely reasonable thing to do you got to be a little smart about it and how you can figure your topics um so it's persisted for as long as you need it but you can do that okay so I'm going to store this data in Kafka um so I'm not going to use Kafka connect I'm going to use the producer API but another requirement that I had at this point was that I wanted this process of adding this metadata updating this metadata uh to be as lightweight and simple as possible from a user standpoint and I am the only user so okay we're gonna we're going to do what I want um so to make it even easier to add this data and update this data and get it into Kafka I don't want to touch my computer right so let's get some help from t so at the end of this I want to receive alerts on my phone and I'm going to receive that on my phone through an app called telegram since I'm going to be getting data from telegram I figured why don't we just use telegram to write the data to Kafka also so if you're not familiar with telegram it is a messaging app like WeChat WhatsApp um but the cool thing about it is that it offers a really convenient messaging API around it um and the ability for users to create Bots okay so you can create a bot you get an API key with which you can um authenticate that bot and write scripts for how people interact with that bot what that bot responds with and what it does in response with it so they have a really cool python telegram bot library that you can use again to write the script to deal with that and so I could write a script to process any incoming messages which and then could produce data to Kafka pretty cool so for any telegram bot you're going to follow pretty much the same structure uh you're going to define a highlevel conversation flow that lays out what are the stages of this conversation what are people going what is what is it going to do in response to any messages it gets um the different conversations it can go through and basically you're just creating a finite State machine of like what it's doing in response to the different stages of this conversation and there's a message Handler a function that then at each stage can handle the incoming messages so going into one of these functions these message handlers so at this case at this point I've already I the bot has already prompted for the low moisture threshold I as someone communicating with the bot has given it a low moisture threshold the bot is extracting that from the user input storing it temporarily in a mapping in its state and then it's going to pass that on to the next stage in the conversation and I'm going to give it the high moisture Threshold at that point so we're going to do this for every stage of the conversation and I'm going to collect all of the data that I need according to into that schema okay and once I've collected all of that data according to my handy dandy schema I can then write it to Kafka and at this point we can follow the same process that we saw earlier um I'm going to create a basic kfka producer it's going to know how to serialize this particular object according to that schema that we have and it's just going to package it I have some helper functions in there to translate between that dictionary to the house plant sort of object and we produce it into Kafka pretty standard so this is what it looks like on my phone um so it has you know some commands that I can give this bot so I want to update this plant and then I follow through with all the information that it requests we confirm it and it's going to produce it to Kafka for me so this is pretty great I don't have to log into my computer to get this data into Kafka just right for my phone and so if I want to add some plants or tweak a couple of the low moisture thresholds I can do that pretty easily from my phone but that's not good enough is it I can add metadata I can update metadata great but what good is adding new metadata if I need to go back into the Raspberry Pi script and manually update that awful hard-coded mapping of sensor ID to plant ID right I still have to go do that so rather than have that be hardcoded in the collection script I figured okay I can make this a little more event driven and I could get that data from Kafka right and so I need to get that sensor mapping into Kafka all right so we're going to collect a little more data here again with telegram so this is a pretty simple schema as well just a mapping from the sensor ID to the plant ID and we're going to use the sensor ID also as the key of these messages and we're going to follow the same process it's a much simpler conversation Handler uh because there's only two pieces of information to collect and here um it's going to prompt me for which sensor ID I want to update and this can be hardcoded this is never going to change and so it's going to present that finite list to me up front and I can put in the new plant ID that I want to to assign that sensor to and cool so that is in Kafka now I just have to change my Raspberry Pi script to have uh to be more event driven to get those mappings from kfka and so we need to get the data out of kfka similar to producing data the considerations that you're going to go through when you figure out how you're getting data out of kfka there's a lot of similarities right you're either going to use the consumer API or CA connect um the consumer API like the producer API it's flexible available in a lot of different languages um you're free to do whatever you want with the events at that point um if you're looking to move data somewhere to maybe a database or some longer term data store then kofka connect is going to be a little more useful in that regard but here we're going to use the consumer API that made sense so I'm going to add a consumer to the beginning of that script so before it starts to collect any data from the sensors it's going to block it's going to build up a full mapping with the consumer of four sensor IDs two plant IDs so we know that we every sensor has a plant that it's mapped to and then it's going to go through that process of looping every 30 seconds and at the end of that Loop it will try to pull for new updates new mappings okay but there's a couple things to keep in mind here so if that application that script were to crash right and we bring it back up which happens then that's fine um that consumer is going to pick back up where it left off because in Kafka the consumers by default are going to commit their offsets they're going to keep track of where what message they've last seen and where they've last processed all right I don't want that though right because if I only have four mappings in that topic and we crash and we come back up well I've already seen those four mappings I'm not going to get them again right so I want to turn off that functionality so in this case I'm going to use something called a non-committing consumer um so to actually enable that you're going to use two configurations first you're going to set enable autoc commit to false so we're no longer going to store those offsets and then in the event that it starts up and there's no offset where does it start and you set that with auto offset reset so we're setting it to earliest so when the consumer starts up it doesn't have an offset with which to start at and it's going to start at the beginning of the topic the earliest spot in the topic and it's going to start basically aresh build up that four um four entry mapping and we can hit the ground running cool so at this case when I update data mapping within 30 seconds we're going to see that populated which is enough time for me to move the sensor from plant to plan and rearrange some things so cool this does what I need it to do that's great so now we just have to actually process the data make sense of it right so to do some stream processing within the kfka ecosystem you have options isn't that nice um you can use the consumer and producer API is sort of the OG way to do it um this is great again available in whatever languages you want it to be available in but to use it for stream processing it's complicated okay especially for State fullstream Processing because you're going to consume from Kafka do your processing your transformations in memory and then produce it back to Kafka but okay what if that instance goes down what are you doing with your state where are you keeping it I don't know if you have an answer to that okay um because it's just an annoying thing to think about for stateless processing absolutely use the consumer and producer API that makes sense but for something stateful like this you should probably use one of the next two things so moving up in the sanity and ease of use is kavka streams so this is a Java and Scala Library purpose built for stream processing for stateful stream processing it is really good at it it takes the hassle of managing State off your hands um and gives you some cool um DSL to play around with so the cool thing about this is that it's built on top of the consumer and producer API so you get a lot of the cool things that you get from consumers like scalability um and the fault tolerance the high availability of these instances and so you can spin up multiple instances of a kova streams application and it's going to parallelize the processing share the input load and process in parallel which is pretty great and if it's a stateful pipeline it's going to commit that state back to Kafka so that if an instance goes down and we redistribute that processing to another instance it can get that state from kfka pretty easily all right so you never lose that state that you need and so to make it even easier we have ksql DB which is a SQL wrapper on top of cavka streams and this is what I'm going to use uh for this project just because it's also built into the conflict Cloud console so it's pretty easy for me to manage so how do we actually process with Casal DB first things first get the data into it and you also have to decide how you're representing the data so there's two main constructs we have tables we have streams streams represent ongoing series of events tables on the other hand are going to represent the latest value for a given key so let's look at that metadata first um this should feel very familiar it's just a create table statement we want to represent the house plint metadata as a table and so the most important thing is where are we getting this data from house plint metadata is the topic where that data is currently residing and the value format is Avro we spent time creating that Avro schema and it's going to pay off here because Casal DV as a consumer is going to look at that topic find out see that it has an avver value format look up that schema in the schema registry and use that to parse all of the fields and the types for this data so we had I don't know like eight or 10 fields in this schema I don't have to manually type all those out it's going to parse those for me and the only one that I have to specify is what field is serving as the key because we want to keep track of the latest value per key right so once I write this query I can execute it within the conflent cloud console and it's going to start bringing in the data from this topic into to this table for the readings data every reading message is relevant in this ongoing series of events right we could potentially trigger the processing off of each of those events and so we're going to create a stream it's going to feel very familiar to the create table statement we just saw and again we're going to point to the Kafka topic specify the value format and it's going to parse those fields and types I'm going to specify the ID field just for consistency and now okay those inputs are there let's actually enrich these two data sets with one another um because we need both of them to actually make sense of the readings right so this is a wall of text just uh doing the select statement first I'm going to extract all the fields that I actually want as output and we're going to do an inner joint just to make sure we have all of the data that we need for the stage of the processing and so if I took this select statement and executed it as is this is something called a push query all right executing this is going casq DB is going to push the result set condu conduct this join and P push the result set to me as output and then stop everything that's in the topic is going to Jo do the join and stop which okay and it's just going to be output that's not really useful so I want to persist this data somewhere and so I'm going to attach this to a create stream statement and so we're going to give this output a name the enriched house plant readings I'm going to say which topic I want to store it in and of course we're going to have that Avo value format so if I execute this great it's going to run that join on everything that's currently in the topic and persist it to that stream cool but it still stops and so no new data is actually going to be Jo uh we're not going to do the join on any new events and so to do that I have to add emit changes to the end and so now when I run this whole thing uh it is now a pull uh sorry this is a push query the other one's a pull query this is a push query and so everything is going to be pushed to me as output every time a new event comes through on the top we're going to do this join execute it and put it into this stream for us to use so now I have all the data I need and we need to start making sense of this information so the overall goal here is that when my plant has enough low moisture readings indicating that it needs to be watered I want to receive an alert on my phone I'm collecting this information every 30 seconds though if a plant needs to be watered I do not want an alert every 30 seconds that's just ridiculous um and before you ask I'm collect this data at that interval just so that I have historic information on my plants because I am ridiculous and I really do like my plants so just have it um and so I don't want to receive an alert every 30 seconds so okay maybe once every six hours that's good right if I'm out at the office um if I'm traveling or something that gives me enough time to either get home and do something to that with that plant or tell my partner to go water that plant right and these sensors aren't perfect they're a dollar right they could have been nicer and so I can get false low or high readings but the general Trend over time is that yeah the moisture level decreases that makes sense um So within a given six hour period it's probably good enough with these okay sensors uh if I receive one hour's worth of low readings right so if you do the math there from receiving alerts uh messages every 30 seconds within a six-h hour period I want to receive at least 120 low readings within that period before I decide to send an alert okay so this is the that I wrote to achieve this all right a few key details to focus in on the first thing I'm going to do is I'm going to set up this query to window not it aggregate the data over non-overlapping six-hour Windows okay so when an event Flows In we're going to bucket into the appropriate window within that window I want to count per plant ID per window whether that moisture reading was lower than the low moisture threshold if it is we are going to update the state associated with this window with this plant ID increment that and if we get at least 120 readings great we are going to Output a result so a couple things to note here because I think it's pretty important um it's easy when you look at SQL to forget that you're building a streaming application so every time uh when we're window and we're conducting an aggregate the output is going to be a table right because within um a particular window we're going to have multiple if multiple events flow in we're going to have multiple potential uh results right so we want to keep track of what the most recent value is and have that output so as an event Flows In okay we're first going to bucket it into the appropriate window and group it by plant ID and then if it meets that threshold if that moisture threshold is lower than the low moisture threshold that we have we're going to in Access that state and increment that counter right and if we hit 120 cool we can output something but you might be asking rightfully so what happens after you hit 120 readings are we going to out output every 30 seconds then CU it looks like we will but we're going to add a little bit at the end here emit final so before we used emit changes emit final is sort of the counterpart emit final says wait until this window closes before you output anything then assess the threshold in this case count above 120 and output if necessary right so this does what we needed to do cool and so that data is in our um house PL low readings topic ready for me to do something with it okay so we just got to get this information out of Kafka so how do we do this well Kafka connect okay um so with our telegram bot we um we were given an API key and with that I can see at a high level what conversations are happening with my bot it's only me talking with it so I have a conversation ID that I can uniquely identify my conversation with that bot using that API key and the conversation ID I can build up an endpoint that I can send information to and receive it on my phone so conveniently there's a cka connect HTTP sync connector that is fully managed on conf cloud I don't have to touch anything and for its configurations it just needs to know which endpoint we're going to send data to and which Kafka topic it wants to pull data from right I'm going to add in a little bit of regx there to extract the field that I actually want to send to the conversation and once this is started up it's going to see all the messages in that topic every time one flows through it's going to extract that data and push it to my phone with an alert like this right just to guilt me a little bit so I'm pretty proud of this pipeline it's a little ridiculous but it solves a real problem that I had have it's an ongoing thing um but what do you get out of this you sat through this hopefully you had a little bit of fun of the absurdity of it but you learned a bit you did I promise um you saw some of the ways that you can get data into Kofa using the producer API or cka connect we were introduced to stream processing if you haven't really thought about it before with Kafka um how would to do this within the Kafka ecosystem with the consumer producer API Kafka streams and especially k equal DB and then you saw how you can get data out of Kafka you know using Kafka connect toward the end and how to use the consumer API to make a script a little more event driven so every streaming data pipeline you're going to build from here on out with CFA is going to use some combination of those three things prove me wrong I dare you um and so now the next time you think about it you at least have the tools the awareness and hopefully you can hit the ground running and I hope you will have you considered bringing kfka into your own home you should it's really fun um so regardless of where you are on learning kfka or um any any of these tools I have a lot of resources here on my link tree um I have a source code linked to uh for this project if you want to take a look at it and if you're really uh inspired and want to do it at home and then I also have a link to confluent developer which is our portal with a ton of uh courses ranging from introductory stuff to more advanced things so I hope you'll check it out and I know I'm at time I will be up at table two I believe so if you have any questions if you want any stickers if you want any Costa socks I have all of those things today and I'm happy to hand those over so thank you so much I appreciate it thank you [Applause] Den