sfscala.org: Paul Kinsky, Getting Started with Akka Streams

sfscala.org: Paul Kinsky, Getting Started with Akka Streams

Recording: sfscala.org: Paul Kinsky, Getting Started with Akka Streams

all right so getting started with akka streams that's that that's my that's my intro it's not I'm not a comedian I don't have like well practiced right so first Who am I Paul can skim a platform engineer nitro we work on making a document smarter PDF slightly less awful to deal with great company but we didn't pay the expense report for the pizza so I'll keep going right so streams that's what this talk is about why should you care first they're everywhere you've got real-time streams of data you've got stuff like streams of log data streams of data from Internet of Things devices sensors streams of data from your machines you have real just big data anything that is too big to fit in memory is necessarily processed as a stream right you have a stream of chunks that you process one after another and of course you have streaming media just bulk transfer of data you have like Netflix your Spotify etc that's multiple billion-dollar companies that just like stream data so it's everywhere TCP streaming data like react that new Hadj a JavaScript framework it works with streams under the hood streams are everywhere it's they're sort of what are streams right two types of streams that people really use distributed and local first distributed there they're really cool it's not what this talk is about you have spark you have storm training models based on real time streams of data and then just of course Big Data so and then local streams right where you want to have more granular control so I say you have complex logic right the first ones sort of based on something we are running in production or gonna be running in production soon but say you have like a really granular flow of data you want to pull messages down use them to fetch things from s3 do something to that thing you just fetch from s3 upload it create an event you can't really do that on sparks you need to control how data flows at like sort of a more granular level or again granularity of control right so I just said TCP is streams so instead of when you get a request framework taking the whole thing into memory and then processing it you can get the header and the rest of the request is a stream throw away that stream and not consume it if their headers malformed so that would be really helpful say it for getting DDoS and of course building distributed systems right so you have your sparks your storms you can't just like put the spec for that on top of an x86 chip and expect it to run someone needs to build those and stream processing systems are a great way to build those so all right here's the traditional stream model right you have a publisher you have a subscriber you send data from the publisher to the subscriber there's some choices with that you can either push or pull data this should be pretty familiar terms right push the publisher controls this initiates the transfer of data so that what happens is data shows up at the publisher it sends the subscriber as quickly as it shows up subscribers has to handle whatever data is sent down the stream and with poll that's like the opposite the subscriber controls the flow of initiates the flow of data so you had have like say polling basically the publisher would go hey or the subscriber would go to the publisher say hey give me data publisher would send it downstream etc and that's if the publisher doesn't have any data it doesn't have anything to send downstream it gives an entity response so you've got a sort of trade off like two by two square here right if you're pushing the subscribers faster that means you can just publish as much as you want downstream as fast as you want and the subscriber will always be able to handle it because it can consume data faster than the publisher can produce it but then the publisher is fatter the producer is faster you're going to get a buffer overflow or you're gonna have to drop data right you know what buffers are bounded or working with machines they're they have finite memory space so you can't just buffer data forever if the publisher is consistently faster than the subscriber you have to drop data or you're just gonna crash other hand pulling right you're pulling from upstream if the subscribers faster that means you just need to get network congestion right you're gonna be pulling for data repeatedly it's not gonna have anything you're just gonna clot you're gonna have your server fall over and of course if the producers faster there's always data when you go upstream to get some all right so reactive streams right that's this is the standard on which akka streams is built it's not the only thing using it reactive streams uses it a few other initiatives engineers from like Twitter and Netflix typesafe we're all involved in creating the standard so the whole idea is that you have bi-directional flows of data downstream data gets sent downstream and then demand is sent upstream right so the upstream sends data or the publisher sends data to the subscriber when and only when there is outstanding demand right and the subscriber signals demand upstream these two aren't necessarily coupled so you could have data or demand sent upstream by the subscriber then ten minutes later the publisher has something to send it can fulfill that demand so it's sort of decoupled it's very message driven it's entirely just based on messages sent between publishers and subscribers so let's look at how that works wouldn't the same subscribers faster publishers faster so subscribers faster the publisher always has surplus domain demand to fulfill that means that it can always just send stuff downstream it's essentially the push model if the subscriber could handle it if the subscriber can't handle it if the subscribers slower than the publisher that means there's no outstanding demand most of the time so therefore when the subscriber sends demand upstream the publisher will basically immediately fulfill it that's essentially the pull model because you have data being sent downstream in response to demand being sent upstream right so so essentially it's switching dynamically between push and pull based on the relative speeds of the subscriber and the publisher which is really great so you it's basically back pressure that's another term for it so if you're dealing with flows of data where this relative speed of different components changes you can on the fuel on the fly switch between pulling and pushing on each leg publisher subscriber connection which is pretty nice but under the example problem right that's that's the introduction so transaction let's say this is stolen from like the second chapter of the book I'm working on act streams in action the idea is say you work for a bank there's a disaster middle of the night you get woken up you've lost all your account balances right everything's doomed but ray of hope you have the transaction log so you can't you have to like build a program that can run through the transaction logs incrementally consume them and use them to construct the account balances so what would that look like right for convenience and because CSV fees are everywhere to CSV file you have your header type name and amount your data it's either deposit or withdrawal and then the name of the account and the amount in dollars right super simple version of the problem more is like just to illustrate some concepts so let's start by building a source that we can use to get the data at the lines from the CSV file right so sources or their source of data right but it's one of three entities and akka streams that you'll be using most often sources flows and six sources produced data flows transform data sinks consume data but the key concept is that they're not actually describing running stream pipelines they're blueprints that you can combine with other blueprints that you can create you can come do so across threads you can materialize them to create running stream pipelines multiple times their thread safe they're immutable they're type they're all that good stuff running streams not so much but fortunately we don't have to deal with those until runtime so our source so we're working with just like some Scala utilities because it turns out it's actually like a non-trivial problem to take chunks of bytes and sprint to split them on new lines does what if the new line what if the what if we need to create lines that are across multiple chunks of bytes you need to create a stateful stage you can't do that knack of streams it's more code than I want to introduce right now so we're just using the iterator of string from just one of the main Scala packages so what we do here we create a function that creates our iterator and we create a source and remember sources are describing the pipeline they're not a running pipeline so what it needs to do it can't take the iterator directly iterators are mutable you can only use them once it takes a function that creates the iterator so you can think of the source as like sort of a composable factory right for a part of a stream but Factory is a dirty word don't forget I said that so here we have a source of lines it's just wrapping an iterator and we can use run for each what it does is append a sink that runs for each element some function which passing in print line and you can see here if we run it on lines on our CSV file yet typename amount deposit withdraw etc the lines I just showed you so that's that's okay right we've done the first step we've got it to lead to the point where we can work with it so we need to actually parse each of those lines it's a comma separated value file common delimited lines right so what we do is we map over each we split it based on commas turn it into a list now this is actually a flow right we're describing a transformation pipeline here so we already have a source of data we want to create like a pipeline that we can bolt on to it that runs our transformations so this is it takes it's a starting with sort of the identity flow that's what a flow dot apply with a single type Ram there gets you just a flow string the string that applies no transformations just passes things through then on top of that we're mapping we're splitting each and so you can see here if we run that with run for each again you get a list of type name amount deposit John Doe 200 withdraw John Doe 50 so that's more parsed right this is sort of another step along getting it to work so now you can see we have sort of a mix of header and data so another thing taken right from the Scala collections framework dropped right so we create another flow pipeline that does the same thing but first it drops the first line again we append that pipeline to the source of lines run it for each by appending a sync and then we just get lists of data so now that's we don't want to work with that right it's strings its lists we want case classes so we're gonna do something like this we have transactions their transactions either deposit or withdraw give an account in the mound and etc just a normal case class so what we're doing here we're using collect again from the Scala collections framework well the same concept at least and we're matching on either a list with a first element being deposit and then the countin amount turning the amount to int or withdraw on the same things now you should note that if an exceptions thrown here when you do to end its will fall over you could make a more better apply function for deposit and withdraw that actually parsed those lists or directly parse CSV lines but outside the scope of this talk so when you run lines through row was through parsed and then append the for each sing to it that prints each line you get a deposit object to withdraw object again right as expected thanks then let's build a sink that will apply the transactions first we need to know how to apply a single transaction we're basically gonna be folding over a stream of them so you have a state you have a transaction if it's a deposit you update the state which is just sorry a map it's just written to integer map of accounting to balance and you well yeah this is just basic map manipulation and then you just create a sink that folds over it so what this does is it starts with that initial state that empty map for each transaction it applies it with a five transaction and then when it eventually finishes when the upstream stream completes there's a future that's yielded when you run this that then completes with the final map so you can see here we create a parse and apply that's a sync so what we're doing there is we're prepending the rows and parse flow to apply transactions we're using two mat here to keep only the right materialized value that's the map of string two integer that's our final final account balances so you create a pipeline that's a fully connected stream processing graph again using two mat and that every Inlet and outlet and that is fully connected right there are no sinks that have empty inlets there are no sources that have empty outlets so this can be wrong so you do that that gets us our future we await it you get a map of John Doe 250 because if you remember we deposited $100 withdrew $50 that's you know trivial but correct so so far this is pretty boring right you can use the collections framework to do exactly the same thing it's like a list of two things that were running something over a parsing etc so let's let's work with a million lines CSV file instead right so here we're creating a source by concatenating a header row and then the just a deposit by Jane Doe for $1 a million times so again we create our pipeline we run that and is expected you get a million things now it's a good time does anyone have any questions all right so every source sink and flow has materialisation type right a materialization the materialization type is something that you get on the side once when you create it so if you're say creating a source from bytes from a socket you might get like a closable from that as at materialization time that would let you close the socket in this case with our fold sink we're getting a future of the value it creates by folding as our materialization type and to Matt instead of two it allows you to pass in a function that just takes the left and right materialization types and chooses which one to keep keep dot write is just a function that takes the rightmost of those two types anyone else right so right keep that left is just a shorthand for keeping the leftmost element of a two element tuple oh sorry right so let's look at some actual types here for a second so let's look at the sink that we created parse and apply it's tiny right so actually it would be better to do this in the rebel right so this is the type of the sink you can see it's a sink with two type parameters string and a future of map of string two ends so that's the materialization type that's the future is something that you get when you're materialize this sink because it's consuming a bunch of transactions applying them to some initially empty map and eventually completing the Associated promise with that future to return the eventual result when it completes so that's included in the type parameter and so let's take a look at lines right that's the source of lines so that has a materialization type unit and when we're combining one something with materialization type the future of something and one with materialization type unit we need to choose whether to keep one none or both of those materialization types and keep dot right is saying keep the rightmost of those materials materialization types when we're appending a sink to a source the sink is on the right the source is on the left because left to right right so keep dot right is just saying keep the future of map of string to end and not the unit there's also keep tup both keep left keep dodd neither does that answer your question all right so so we created a pipeline right something that we can use to complete like the semi trivial task of parsing sorry parsing a CSV file incremental e consuming it and I running something across running something across each elements each row and shown it does work with a million line CSV file I can do this in the rapid if you want to but I found that it's wise not to tempt the demo gods right so we've created our pipeline let's see if we can do something completely different with the same pipeline say take a WebSocket push messages describing transactions up that WebSocket and get the current state back so we need to change a few things to do that first we need to go from messages to strings and back so what we're doing we're using a KH TTP and collect again because WebSockets support both binary text messages so we're only taking the text messages grabbing those piping them through all rows because this time we don't have a header row we're just using CSV over WebSocket because we do crazy stuff like that parsing it again and here we're using scan right instead of fold what scan does almost like fold but instead of just out omitting the last element it omits every element so every update to the state will be emitted sent downstream so what we're doing with that then is we're using those States to create new text messages and this is being sent via the WebSocket so let's check that out the surrounding code is also in here so you have a really simple DSL that'll be skipping the describing so you just have any get with no path we'll just serve up this resource this HTML page that is the front end anything two WS will handle every WebSocket message with this flow parson apply WS so apparently the tooltips are not zoomed in on that's just the flow that we showed in the PowerPoint and then we're just binding to a localhost port 9000 so I've got this running in the background and the UI is leaves much to be desired it's using the console for logging everything sent by the WebSocket because I'm a back-end engineer and that is how I consider UI right so let's have again Jane Doe deposit a dollar so you can see the resulting state is sent back down the WebSocket so just keep doing that increments take someone else thousand dollars why not and take $3,000 or take a few thousand dollars so this is not what you'd use for your bank right you can get negative negative account balances really easily it's just to show how you can do it work with streams of data right and you can do more complex things with WebSockets you could it very easily integrate something that works with futures like if you have say if you currently use futures for interactions with your no sequel store with your database there's a map async that takes a function from A to future B instead of a to B you can like specify parallelism you can do map async unordered if you want to remove the order and guarantee in exchange for higher performance if you get an occasional slow element there are just a ton of Combinator's that you can use more so even than in the Scala collections library maybe to transform streams there's transform specifically sources and flows then you can also create much more complex things using the graph DSO which is outside the scope of this talk so say you can merge multiple streams together you can broadcast one stream to multiple consumers you can our balance a one stream among multiple consumers if you want to add parallelism a lot of stuff you can do rights and you can use Kafka for example there existing libraries you can use to stream to and from Kafka to post grass etc so uh contact info nitro careers I'm email my github does anyone have any other questions cool oh sorry I should have explained that in more detail right the question was it seems like map and collect or are the same they almost are map takes a function collect takes a partial function so what collect does if the partial function is not defined it drops the element so if we were to send like an empty line up or WebSocket for example collect would see that there's like an empty list after you split that on commas and that would just be dropped whereas if we use map with a partial function look any element that was malformed would cause it just to fall over with a match exception that's a I think that's all the questions