sfscala.org: Paul Kinsky, From Websockets to Kafka with Akka Streams
Recording: sfscala.org: Paul Kinsky, From Websockets to Kafka with Akka Streams
okay cool the goal of this talk is to show how you can use a cast reams to compose stream processing well composed different stream processing components and I have not gotten my water before that well right goal this talk is to show how to use a customs to take messages can receive via web socket from web socket connections to a server and publish them to CAF cut before we start let's go quick overview of akka streams right I won't be getting into the details of accra streams how to use it how to use different features this is more of a talk designed to motivate you to go out and learn about akka streams by showing how you can use it to compose well more or less anything stream processing related so a few examples here right it's like Legos for stream processing I just came up with that today I may or may not work right so you can take Kafka you can stream messages to or from Kafka you can stream messages to or from a WebSocket connection how you can do high-level Network I oh you can have streams of HTTP requests to HTTP responses you can define a server that way you can do low-level network i oh you can send streams of messages directly over tcp between servers if you need to like if you need to skip the overhead of using HTTP itself some people do you can use files local on s3 pretty much that's in the same way you can stream bytes to and from them and if you can connect you all of these things to the input or output streams of processes so if you wanted to like stream a file from s3 into the input stream of a process and then stream the output stream of the process to another file and s3 you can do that really easily it's not what this talk is about but it's really important to a just sort of a hammer home that you can like use it as Legos for stream processing so so our goal tonight this is quite small let's see if I can trying out a new a new presentation framework so can everyone read that part of the bottom can some people read that part at the bottom all right sorry them right right so what we're doing tonight we're creating a server it will consume events via web sock it will publish all of them to a single Kafka topic so at that point you can do whatever you can use that as input to spark to whatever your big data real-time data analysis platform of your choices you can just read from that you can basically use it with anything Kafka is a very good glue for different systems in the micro service architecture so in case you can't read that bottom line that's the case class that will be working with it's an event there's a client ID there's time stamp there's nothing else it's pretty clearly meant as a placeholder you can put like messages there you can put whatever data really that you wish to collect from the front end so we'll be working with that event case class for the rest of tonight and there's a bit of boilerplate here that you really don't need to read all of basically what this is on the companion object of event first very simple you have the JSON serialization and deserialization by a format that's why I play JSON it just uses macros to create some implicit classes that will convert instances of event to and from JSON and then here you have a slightly more verbose serializer and deserializer used for Kafka all these really do the serializer turns events in two bites the deserializer turns bites and two events will be using those to publish from or publish to and read from Kafka later in the stock so as always there's boilerplate riot implicit boilerplate first an actor system that's as from the top before it's a knack actor system it provides actors as a unit to run computation in a distributed environment they communicate by a message passing then there's an execution context that's just the system dispatcher that will be used for mapping over futures and registering callbacks and futures finally there's the materializer so that just uses the actor system the implicit actor system to turn basically blueprints in extremes representing stream processing graphs into running instances of those stream processing graphs using actors created by the actor system that's that will be all that all be in the background but if something looks like magic it's important to remember there's a lot of implicit more heavyweight stuff that it's running on and that's what that is right so let's zoom out a little bit right sorry the materializer it takes a description of a stream processing graph that's a series of nodes that communicate with each other via streams right that stream data from node to node B and it turns that into a running instance of that graph created using actors by analogy it's like an execution context as an execution context is for futures materializer is for a cust reams you don't need to understand the details of it it's just something that needs to be in the implicit scope for most of the stuff to work behind be happy to discuss it more later but if you're willing to accept that it's magic and then let me show you how you can use the magic to build cool stuff very concisely the idea is that that will motivate you to learn more about it but can I'll definitely have time to answer questions after and and during to of course flip so this is getting started right and most of these parts of this talk will as down there have these little ascii art diagrams and those are showing different components of the stream processing graph that we're building right so this code it creates this a stream processing graph component that consumes events right that's the error going into it it consumes a stream of events and publishes them to Kafka so the specifics of how we create that first we use reactive Kafka it's a library that uses akka streams to communicate with Kafka or reactive streams rather which is the standard that akka streams is built on we have our Kafka instance it's running on docker machine locally that's just hard coded because I can and we use that to create a sink so what that is is something that consumes a single stream of messages of type event and when we run it it doesn't the side effect of running it is unit right so produces an instance of type unit a que voy de que nothing we can we can ignore that for now so what we're doing is first we're creating a flow that just does nothing right it's the identity flow it's a from event to event then we're mapping over that we're wrapping every event in a producer message finally we're sending that to a sink that we create using the CAF go client and some properties that we pass in right so there's the IP of the the port of the local Kafka instance there's a topic name that's the that's the topic that we'll be publishing too and then there's the serializer that we created in this giant glob of serialize deserialize boilerplate all the serializer really does is take events turn them into bytes those bites are then published a Kafka which is a distributed log framework or log service so at the end of that though what we've created your defined is a blueprint for a stage inner stream processing graph that takes events and publishes them to Kafka will have a lot number of these asked ER diagrams they're designed to be composed right so if you have another source of events we can just sort of plug that together with this sink without needing to know every detail of how the sting is implemented that's really the key to the whole like compositionality of it the whole like legos approach right so we're going to do now and this is this is a bit hacky to be honest we're going to use a source queue to create a single running stream that are multiple streams can publish to for every WebSocket connection will be getting a stream of WebSocket messages those will all be publishing to the source queue which will be connected to the CAF go stream or cafes think so a source queue right it's very simple very simple trade it's just something that you can offer an element to it'll either reply with the future false if it's not able to accept that cuz it's a queue buffer has been overrun or a future of true if it accepts that so right here we create a runnable graph by combining our source or Q source right given like 1024 as the buffer size and an overflow strategy that we can gloss over for now and we're combining that with the dot 2 with the CAF cos Inc that we created on the last page so you can see here it's a sink that takes events and this is a source that creates events so we combine them now there are no hanging inputs or outputs that's just a runnable graph that we can create everyone with me so far alright cool so yeah so then we run the runnable graph right and we get a source queue as the output of running it so it's graph of types or skew of event that means that when this blueprint right this graph is just a blueprint defining it hasn't actually caused much computation to occur when we run it we get a source queue of events that's how you can get things out of materializing a graph like futures if you have any way if you're like folding over a stream for example so now we have a source cube because we have actually run it using the materializer we define materializer we defined earlier and here's the shape of the stream processing graph we just created right events go from the source q2 Kafka and we can just use the source q to NQ events and to be sent to Kafka so now we need to actually write to that q right so we do here we define a queue right or abstract in terms of type T because we can and all that does it takes a source Q of type T uses creates like an identity flow using flow of types flow of tea uses map async to offer elements the queue and then sync dot for each to just run like a function on the result of every offering right so if it's false that means that it was unable to give a in queue element on the queue and it print uses print line to just log in air and if it's true we do nothing because that means the message got onto the queue yay we've succeeded so in this case it's an element of type T that's sent to a source queue of type T the store sku provided from outside when you're creating the graph stage right so now we have our stream to cough stream of messages to Kafka we want to start dealing with stuff from the WebSocket right that we're going to open a lot of web socket connections in the near future so first we want to parse the messages so we have a flow from messages to events that takes just text messages from and parses them as JSON and attempts to transform that JSON into an event and will throw an exception if that fails because this isn't production code we can do stuff like that so this flow right it's just a component it's a sort of equivalent to a function not really but you can think of composing flows as function composition in that you have a flow from message to event you have a function for message to event the flow runs over multiple elements the function would run over a single element there conceptually equivalent but not implementation wise so we've defined this other component let's get a bit more complicated right so we're going to bolt that on to parse messages onto the cue writer so all that does is it parses messages that are coming in from somewhere it writes those messages to the Q by Q writer and then that is right here well i can't select just part of the line so it doesn't really work but right here anyway everyone wave your hands around right here that's a sink so that's a sink that consumes messages and store stop maybe is a source that may or may not produce a single element here we're not going to produce anything we just have this here so that the WebSocket connection doesn't close we need to keep that a promise of a potential elements there so then we combine the two float up from sink and source right as up here we take a sink in a source that aren't perhaps aren't related at all we combine them sink the source is the input of the flow the sink is the output right this is what we're going to handle every WebSocket connection with every WebSocket connection is a flow for message to message message is consumed by that flow are sent from the web page via the WebSocket connection to the server message is produced by that flow here in this case we weren't have any but we don't need to tell the compiler that will be sent back down the WebSocket connection to the web page right so now we're going to have to have a flow from HTTP requests to response because we're going to have multiple web socket connections right this isn't an interesting talk if there's only one WebSocket connection open under private so we're using a bit of basically magic from a khg TP to define a routing dsl domain-specific language so what this is doing if we get a get request with that being the end of the path we'll just serve up tests on HTML from the local resources file if we got to get requests to slash WS WebSocket we will handle that web socket message with the flow that we created earlier right here so every message will be parsed and then sent to the source queue to be sent via Kafka and then we'll just not send anything down the pipe so we don't need to write so now we run that using a KH ttp what we're doing here is saying we want to handle every request to port 9000 on localhost with this routes flow right this flow from request to response that like at a really basic level that's what like a server is right it's a it takes HTTP requests and yields HTTP responses and you probably had it explained to you that way in some basic class but then you can't really implement a server as an actual flow there's a bunch of other stuff you have to do but with a co streams you kind of can simple example that this is you can see down here there's a ascii art drawing of the routes flow so finally tested HTML right I didn't actually introduce that it's i'm not good at javascript and not go to HTML this is a very simple page all it does is generate a random client ID open a WebSocket connection to localhost / 9000 web WS and then log to console everything that happens with that and finally every second every thousand milliseconds it sends a events with a client ID and a timestamp to that web socket and logs it to the console so finally because we want to actually see that stuff actually goes onto kafka we're going to listen to cough go with a consumer so all this does it consumes messages from Kafka same local Kafka instance same topic events it also is a group ID right if you have multiple groups they only consume every message but if to consumers of the same group ID the only see each message once that's really useful if you want to parallelize like handle something with multiple instances reading from Kafka and we just then map over it to unwrap the that the rapper that's all our messages have just get the value finally we run it with run for each what that does is it just creates a sink that runs function using for each on every element consumed by it so you can see the shape of the graph down there right starts with Kafka takes events over to sync for each and then runs print line on each of them so let's go to the actual presentation or the actual demo so we already have some messages on Kafka from when I was testing this earlier so you can see what that looks like so right here stop presentation mode I'm going to open up a bunch of local instances so each of these is going to publish events with just its client ID and timestamp once every second and you can see that in the console log nothing in the UI because I'm back-end developer and I don't do you eyes so yeah you can see here every second different client ID / service different time stamp and over here we've got our server running listening to Kafka so every message here is something that's been published a Kafka by our server and then read back from it and you can see that these have a bunch of different like different IDs just mix together now let's let's just throw out like in more ridiculous amount of these because we can right so still publishing we haven't dropped in the elements I haven't tested this with an actual reasonable web-scale number of like 10,000 clients or something only because I don't know any libraries for like simulating a ridiculous obscene number of web socket requests at the same time but if anyone has anything they recommended later I'd appreciate that so that's basically it this you can see the entire example here it's in 88 lines not including the HTML page including the imports so it's we let's recap what we did right we created a persistent stream to kapha bio which we have a source q that we can just getting with a function that returns future of boolean and q messages we created a server or by a flow from request to response and then use another flow to handle every WebSocket messages by a single route there and then we that created another consumer for Kafka to log all those messages in really just like a tiny amount of lines I've implemented things that do like a fraction of this that have taken like three or four times as much code of course every line of code is another opportunity for bugs bugs are roughly proportional to lines of code so anything you can do to make things more concise assuming it doesn't make everything else worse like some of the stuff I've done with scholars even though it's really fun sorry that was mean to anyway yeah so it's really concise you can do a lot with it you can glue a lot of different things together any questions I guess not up just to be like what happens behind the scenes to keep like the WebSocket open because they the requester is to create the web socket and then I think you'd mentioned me there maybe or something how is the magic oh great question right so um sorry what happens behind the scenes to keep the WebSocket open let's let's go back to the example for a second or the page with that for a second so sore stop may be here right if this was a source that just instantly closed it would instantly close the WebSocket connection right what this does source top maybe let me actually show you it from a escala standpoint is a source that materializes up to a promise so when you create it it just like spits out a promise and if you complete that promise with some something it'll admit that and then finish if you complete that promise with none it'll just close if you're never complete that promise if you just leave it hanging it'll just always be open and like not actually completed so that's what source thought maybe does it's just sort of there to not complete because we really want to treat the WebSocket as a sink that just consumes elements but sorted by like the nature of the WebSocket protocol it wants to both consume and produce elements and we have to give it something to consume that doesn't close instantly and that's what maybe is used for does that answer does that answer your question awesome anyone else browse web sockets are cool next question or sorry you asked why i chose to use web socket so when it's a one directional communication when I could perhaps use like oh totally you could really just like post every time something happens but i don't know i mean is it actually beneficial if you want to publish events in a regular basis to use web sockets instead of like HTTP POST i actually don't know i just jumped a web sockets because they're cool and I wanted to do a demo but that's actually a very good question spray I think akka HTTP and spray are sort of converging right like yeah so as I understand it play in a que HTTP are sort of going to be merged if you want like a WebSocket connection in play you'll be using a khg TP in the next version or two theoretically but yeah spray and like I believe they are routing dsl used here which is a the a key HTTP routing dsl is originally from spray right so it should be familiar if you are familiar with spray well I think that's it if anyone has any other questions I find me after you