Scale By The Bay 2019: Colin Breck, Maximizing Throughput and Scalability for Akka Streams
[Music] before I begin the usual disclosure that these are my personal opinions and they don't necessarily reflect those in my employer so that akka streams is is naka streams API is an efficient API and as long as you're not constructing extremely complex graphs it's pretty easy to get impressive throughput but there are situations however we're careful attention to how the stream is actually executed can greatly improve the performance and throughput as well as the scalability of your application and that's the focus of this presentation I'm gonna look at three topics first is maximizing the throughput of an individual stream second is how to partition a stream in order to maximize throughput and then finally how to partition the stream for scalability as well as high availability now this is it's not an introductory talk to akka streams I assume you have some working knowledge of akka streams but I think even if you haven't touched them I think you'll the talk will still be digestible if you are looking for an introduction to akka streams or the the motivation for using akka streams check out this presentation of mine from Scala days or these two blog articles that that motivated that presentation so before I dive in I'll just provide a few examples of streams to kind of motivate the problem so streams can be ephemeral so that might be something like a source that emits a fixed set of elements like a list that you then filter out some elements may be perform a mapping into a different data type and then send those elements to a sink like a file or a database or an HTTP response you can also use a stream to to process the results of a future such as a request to a database or you can also use streams to provide bounded resource constraints so say for a large workflow you might want to limit the number of asynchronous operations that are happening at any one time streams can also be unbounded so consider a durable message queue that receives an unbounded stream of messages from IOT devices application log messages or business transactions it's pretty common to have stream processing applications that to data transformation enrichment digital twin modeling inventory management machine learning anomaly detection and these streams tend to run forever on that unbounded stream of messages acha streams can also be used to run continuous ETL applications and the alpaca project provides a rich set of sources and syncs that will connect different systems and even individual services or devices can have unbounded streams of messages like an IOT device that's sending telemetry over a WebSocket connection which is actually modeled as a as an unbounded stream so whether a stream is ephemeral or durable bounded or unbounded the strategies to improve the performance basically boil down to the following I want to minimize latency so that kind of end-to-end latency of any single message maximize the overall throughput of the entire stream and maximize scalability through efficient use of resources so a reminder of the the three topics that I'm going to cover and just a note on the examples that the examples I provide are all like little toy examples rather than something that's super elaborate and they're actually examples you can play with yourself to see the impact of changing the streaming model that you're using okay the first topic is maximizing the throughput for an for an individual stream and the most fundamental thing to understand is asynchronous boundaries so consider this simple stream with a source and then a couple of map stages and finally a sync encode it looks like the following the source emits a thousand elements followed by the to map stages and the the messages are then sent to a sync that simply ignores them and the spin function that I'm using in the map stages looks like this and it's just designed to waste CPU for 10 milliseconds and then pass the value through okay so back to the concept of asynchronous boundaries acha streams are executed by actors under the hood and you really don't need to know about this or pay attention to this until you want to start improving the performance of a stream operator fusing is a technique that they use to execute adjacent stream stages on the on one actor and this avoids the overhead of message passing across actors to increase the performance so in this example it means that the two adjacent cpu-bound map stages cannot execute in parallel unless we take additional steps and the way to allow these two adjacent stages to execute in parallel is to insert this asynchronous operator which inserts an asynchronous boundary and the stream will be executed in two actors now with message passing across that asynchronous boundary and what this looks like in code is instead of having the adjacent map stages they are separated with the asynchronous boundary with the async operator and this stream will execute twice as fast now just because it can execute those two stages in parallel another way to introduce parallelism is through the map async operator map async can be used to execute a future and then emit the completed result of the future to the downstream and you can set the upper bound on how many futures this stage is executing at any one point in time by adjusting the parallelism of the map pacing so returning to the stream that we just looked at this street this stream will run equally fast if instead of using the asynchronous boundary we capture that spin function in a future and then use map async so this is a very important thing to understand is what's the difference between async and map a sync so we're visiting the original expression of the stream when everything was in one a synchronous boundary inserting the asynchronous boundary with the async operator introduced parallelism by executing the stream in two different actors map async can achieve equal performance but what it's doing is achieving parallelism not through an asynchronous boundary but by executing the work on it in the future potentially on a different thread and these two concepts can actually be combined so so taking that expression of the stream using map async you could insert an asynchronous boundary but to between the two map facing stages and what this would look like in code is similar to before using the map pacing operator between those two adjacent stages now in this case all this does is add additional overhead this doesn't actually improve the performance of the stream but there are cases where you where this can improve performance however adjusting the parallelism of the map async here will improve the performance so if we set it to 2 instead of 1 this will allow the stream to execute twice as fast again for 4 times faster than the original expression of the stream assuming that there's enough CPU and note that that's that's a parallelism that you could not achieve with the async operator itself just because you can't achieve that level of parallelism by inserting it so those are your two tools for for playing with parallelism and so accordant an important question becomes how to tune the parallelism of map async to achieve maximum performance so I already showed how easy it is to adjust the parallelism say from 1 to 4 and encode this looks like something like the following going from a parallelism of 1 to 4 and this allowed to stream to run four times faster but how do I know it wouldn't run 8 times faster if I put 8 so the story is basically for a CPU bound workload like this spin function that I've been using you should set it based on the number of cores that are available but of course you need to balance this decision by the composition of the stream if you have two or three map async stages you need to consider what they're all doing or if your application is running more than one stream you need to balance that trade-off but assuming that you that you only had say one map async stage and all available cores set it to the number of available cores for a CPU bound workload but it also raises the question if it ever makes sense to do something like this set it to a thousand and the answer is actually yes so so far I've looked at this CPU bound workload but consider this non blocking workload that that would simulate say a call to a web service or a database something like that and that is that's non blocking so it creates a promise creates a random number of milliseconds between 0 and 100 schedules the promise to complete at the end of the time out of then just passes the value through again through the future so this stream runs 1000 of those non-blocking calls using the map async parallelism of 1 and this will take you know somewhere between 50 and 60 seconds to execute this stream but with a parallelism of a thousand it'll complete in approximately 100 milliseconds so that's three orders of magnitude faster and it's going to run all of those non-blocking calls at the same time and so that's a situation that you could have if you're calling some you know very scalable web service and you can just run all that work at the same time and you're doing very little work locally so you can basically set the parallelism to what that other service would tolerate so this a site stuff seems really great so let's put it everywhere and this isn't always advantageous as you can probably imagine so consider these three case classes that simply wrap an integer and this stream of a million elements that just you know maps through each of the case classes if an asynchronous boundaries inserted between each stage it will execute twice as slow similarly if each mapping is executed in a future the performance will be equally poor and here I'm using future apply on purpose rather than future future successful so the reason is that the first implementation introduces the overhead of message passing through actors across that asynchronous boundary again and the second one introduces the overhead of executing the future on another thread so you need to use some caution the parallelism can't be introduced blindly the overhead must be justified by the performance gains that you're gonna get there's also an unordered version of map async that can improve performance in some situations so a really powerful feature of map async is that it emits elements downstream in the same order that they were in the upstream no matter which order the future is complete in and this is really important when you need to maintain order and guarantees in your stream but if the workload is non-uniform it can lead to a head of line blocking if one future takes a really long time to complete and if you can relax ordering guarantees map async unordered can be used to improve the performance of these streams by emitting elements downstream as soon as they complete and then starting the execution of the next future and to demonstrate this instead of that spin function I was using so far consider this non-uniform version that's going to pick a random number of milliseconds between zero and a hundred and then consume CPU for that long this stream that runs that non-uniform CPU bound workload will run faster if the map async stage is replaced by the unordered version and the more variation that you have in that non-uniform workload the more dramatic this effect will be and it's especially dramatic if there's a really small number of events in the stream that take an extraordinary amount of time to process now buffers can have a really subtle effect on performance or a really dramatic effect on performance depending on the composition of the stream and they take a few different forms so when when you insert an asynchronous boundary the stream is now using message passing across that boundary like I described earlier and there's there's actually a small buffer of sixteen elements which is essentially the actors mailbox and this can indirectly improve performance when workloads are again non-uniform by keeping the downstream saturated so here's an example of for map async stages that each run some non-uniform workload this stream will complete faster if an asynchronous boundary is inserted between each stage now the improvement in performance here is not coming from the asynchronous nature of that change but just from the fact that there's buffer the sixteen element buffer between each stage that ends up keeping the stream more saturated and you can actually achieve equivalent performance by inserting a buffer so here this is still there's no asynchronous boundaries here it's it still has operator fusing it's executed on one actor but it'll have equivalent performance another place where explicit buffers are really useful and saturating a stream is reading messages from a work queue or a message bus so if there's some messages that are much much larger than others or takes significant longer to process in the stream or transfer from the upstream a buffer is really helpful for saturating the downstream rather than sitting idle waiting to fetch the next message so here's an example of Kakaako consumer and depending on how the stream processes messages inserting a extremely large buffer sometimes can can really improve performance by saturating the downstream if it's non-uniform again but just like we saw with indiscriminate use of a sink indiscriminate use of buffers doesn't end up helping in terms of performance so take this stream with a couple of map async stages if adjacent stages perform relatively uniform workloads then inserting a buffer is not going to change the performance it's just going to consume more resources the performance is going to be dictated by the slowest processing stage and the last topic with regard to the performance of an individual stream is the handling of exceptions the default behavior when a stream throws an exception is to stop the stream but you can define this decider mechanism to have a few different behaviors including restarting the stage and continuing processing or simply dropping whatever that message was and keep going and so the decider allows you to set those rules and it also allows you to inspect the exception and take different different actions when it's thrown so consider the following stream that basically throws an exception for almost every element maybe this is some you know legacy API that you know still throws exceptions that you have to deal with for them like a you know message parsing library something like that so this stream will run two orders of magnitude faster if those elements are filtered out in the stream rather than blowing up the call stack and doing that filtering in the decider and of course there's there's many many ways to filter streams you know here doing it with collect but there's a whole bunch of different stream compositions you can use for filtering another approach and I think arguably a better approach is to model those errors as data this is actually it's especially important if you're doing something like reading from Kafka and you need to you need to manage your consumer offsets downstream you can't necessarily throw away all your errors here your errors need to make it all the way to the end of the stream for handling so modeling errors is data and perhaps even sending those to a different sink for handling can often improve the performance rather than throwing exceptions to the decider okay to summarize maxing the maximizing the throughput of an individual stream want to increase parallelism and the tools for doing that or a sync and map async but you don't want to necessarily increase it too much be be cognizant of your resources and how are you using them and then operator fusing is there for a reason you know don't don't necessarily destroy that performance benefit for non-uniform workloads you can insert buffers and you can also relax ordering in some situations to improve performance and definitely avoid exceptions either use filtering or model errors as data and this is a really important point that I want to emphasize how easy it is to experiment with improving the performance of your streams right we didn't have to write reader/writer locks or introduce new threads or you know pub subs or queues anything like this you insert an insert an async stage increase the parallelism insert a buffer insert a larger buffer these kind of things rerun your benchmarks see if performance improves you don't need to write hardly any code ok the second topic is how to then partition a stream to maximize throughput and I'll look at broadcasts to begin those broadcast is a way to partition a stream by sending every element two independent down streams so every down stream is going to get the same the same stream so in this example the down stream performs a mapping before the streams are then merged back together and this is what it looks like in code using the graph DSL so you set the number of partitions to 4 oops create the broadcast stage with 4 partitions then a merge stage that accepts 4 inputs then define the down streams here that each have one of those map stages that wastes CPU and the spin function and this graph can be run with a thousand elements and it looks like the follows following now just as important as understanding asynchronous boundaries for individual streams is understanding them when you're partitioning streams so this graph broadcast for down streams will run four times slower than just running one of the down streams in a single map stage and the reason for that is it's all running in one actor so to improve the performance it's equally important to apply asynchronous boundaries when you're partitioning streams so that the down streams can run in parallel on different actors and returning to the graph definition achieving this performance boost is as easy as before just inserting that asynchronous operator another way to partition for performance is using the partition operator and this this sends individual elements to one downstream based on a custom partitioning function in this case sending one quarter of the messages to each down stream so it's common to you know exclusively partition a stream based on a unique identifier or you know a consistent hash of an address something like that and that will maintain total order within each partition but it won't necessarily maintain total order across the stream and then how do you pick your partition size what's kind of back to what we were talking about with Matt basing earlier if it's a CPU bound workload you might want to choose your partitions relative the number of cores you have if it's if it's not CPU bound you you may be able to increase your your number of partitions you know to be quite large so in code broadcast looks very similar in terms of graph definition selecting the number of partitions and divining and defining the partitioning function and again taking care when we're defining the down streams that they're that they have a synchronous boundaries so with each down stream only processing one quarter the elements it will complete in one quarter at the time of the broadcast example that we just looked at now a really interesting way to partition streams but it's only applicable to a subset of workloads is balance so a partition element that we just looked at it's going to back pressure when the current currently selected down stream is back pressuring so in situations where downstream workloads are non-uniform and the total order of the stream or an individual partition is not required balance can be used to saturate the down stream using the down streams is like a pool of available workers and it'll emit the to the very first available down stream and it's not going to back pressure until all of the down streams back pressure so this non-uniform spin function will help help demonstrate this so every fourth element takes an order of magnitude longer to execute and then the definition of the balance is similar to what we've already seen with the graph stages again they sync so running this graph using balance will execute three times faster than using partition but of course being able to use balance depends on the workload being non-uniform and also being able to relax ordering guarantees so if you need ordering this this model of stream partitioning is not gonna work now akka streams has a bunch of dynamic stages as well there's too many to detail in this presentation and their applicability is really workload dependent but I'll just give you a flavor of a few of them to see you understand the performance considerations starting with group by so similar to part it's similar to partition in that d multiplex is a stream into a set of independent down streams using a partitioning function but it differs from partition in that it emits a sub flow rather than the more common source or flow but it can be used to consistently partition the stream what it looks like in code is a group I element that takes a partitioning function similar to the partition we already saw then waste some time you know waste some CPU in a map stage it's equally important to insert asynchronous boundaries here otherwise the sub streams will not execute in parallel and then sub streams are completed by attaching them to a common sink or by merging them back together and this this example uses merge sub streams to merge them back together now a few caveats about group by group by is dynamic so if there's a huge number of partitions like your partitioning based on a session ID or something like that it can consume significant resources and they'll be can continue to be consumed until the stream is stopped there's no way to recover them so it's better to do that kind of processing by interfacing streams with actors and then controlling the lifetime of resources by stopping an actor after an idle time Oh something like that and the second thing is if you try and limit parallelism here with the merge sub streams with parallelism and that's less than the maximum number of sub streams emitted by the group by partitioning the stream can deadlock and the split when is is somewhat similar for example it allows splitting a stream into sub streams based on a predicate so this might be something like a windowing function or a watermark and this its effect on performance depends on the size and the number of sub streams you're kind of able to to admit from it and lastly the last dynamic stage I want to talk about is broadcast broadcast allows a stream to dynamically subscribe to a source and new streams can be added on the fly join the stream of messages in progress and this is really useful for subscribing streams and fanning them out to something like say you're using server sent events WebSocket gee our PC streaming something like that but an important thing is that the rate the producer rate will be automatically set to the lowest downstream consumer rate so an important thing to do is to insert a buffer stage to decouple the down streams from the broadcast hub and if the budget if the buffer fills up for a downstream consumer you could just stop dropping those elements perhaps or you can even fail that slow or idle downstream consumer so that you recover resources and that you don't impact any of the other streams okay to summarize partitioning for throughput it's very workload dependant your workload needs to fit the model you can broadcast the whole stream and run independent streams off of it you can partition to divide and conquer balance to have this pool of workers model and then there's these dynamic stages that basically run in depend downstreams okay so the final topic is partitioning for scalability and high availability and this is there's only a very small part of the presentation I'm not going to dive deep at all but these ideas are really really important if you want to build systems with huge scalability and reliability so the first technique is to scale streams by static external partitioning so for example a stream might be processing files from a dedicated s3 bucket or from the file system of an IOT device and as you have more devices or more buckets you just run more streams one dedicated to each and the streams run independently and most likely on eat on different computing resources which provides great scalability and it also provides a lot of reliability because you have isolation across those streams another approach is to dynamically varnish partitioned streams so imagine an ETL application reading from a work queue or a consumer reading from Kafka something like that if this stream in one of those applications is not enough to keep up with the workload even after you've applied all the techniques that I've talked about so far in this presentation you can scale the stream up dynamically running more workers to satisfy the workload for example by scaling the number of pods deployed on kubernetes and after scaling up you know a Kafka consumer is going to rebalance those partitions across the consumer group or if you're reading messages off a work queue like sqs now you have more workers reading off that same queue and a final way to partition streams to provide great scalability as well as fault tolerance is to use aqua cluster charting and the idea here is to let an individual actor run an individual stream or group of streams and maybe that's a stream interfacing with the individual IOT device and then you let akka runtimes scale this to millions sharding those actors across a cluster of coordinating machines this provides tremendous scalability through parallelism and elasticity and also fault tolerance because if one of those servers fails the actors on that server will be rebalanced to another server than the stream that they're running will be restarted and this is a really really is this a lot this is a very useful way of building applications that model entities especially in IOT that incorporates streaming workloads as well as state as well beyond the scope of this talk to get into this but if you're interested in this subject you can see this talk of mine from reactive summit on how to incorporate actors and streams in this manner or this four-part series of articles that I wrote that basically motivated that talk it's worth noting that these articles the ideas in these articles still apply but I think there are three or four years old so they predate acha typed I would definitely use archetypes to implement this kind of thing so to summarize partitioning for scalability and high availability use static external partitioning for static or declarative deployments dynamic external partitioning through consumer groups or work use or dynamically with actors and cluster sharding and what these do is run the streams independently and you let the runtime recover from failures or provide dynamic scaling of your streams okay those are the three topics that I wanted to cover the main takeaways you need to understand the dynamics of your streams and your workloads as well as the external systems that you're interfacing with introduce asynchrony to maximize concurrency buffers to avoid starvation and to provide decoupling among stages of watch out for head-of-line blocking partition the stream internally for concurrency and partition streams externally for scalability and high availability and again I can't emphasize enough how easy it is and how reliable it is to experiment with these techniques run your benchmarks adapt improve performance that's me on Twitter if you want to stay in touch that's my blog I write one article a month there's a lot of stuff on acha streams on there there's also stuff on managing teams and where I think the industry is heading in terms of stateful server lists and these kind of things if you want this presentation in written form refer to these articles I got one minute for questions I think there's a I think there's a microphone okay just a really foundational question let's say you have two operations you want to run in parallel but they both need to complete before you can proceed again and then right after that you have another two operations that need to run in parallel how would you express that with map and map async async' and all that it sounds like you're gonna need you're gonna need bifurcation and merging of your stream but that's that's totally possible so so you express it in a craft so you'd have to use a graph for that okay all right um why is it that keeping the downstream saturated leads to better performance isn't that otherwise it's gonna be sitting idle I think the Kafka example is a really good one that's the first instance that I encountered it we had an extremely large message so even though the consumer itself had a small internal buffer every once in a while there would be this enormous message that took a while to transfer so the stream is sitting idle downstream doing nothing and you're waiting for this message to come across the wire but if you keep the downstream busy while you turn you know do Network IO in the background such that you have this you're accumulating your messages upstream then the downstream can always be busy statistical multiplexing problem question basically right yeah cool Thanks do you ever find there's cases where you drop into custom graft stages to optimize further than the dsl allows very rare and I I can't think of ever having done it for performance reasons it's more for expressing you know business logic reasons not I can't recall ever doing it for performance and and even for business logic is pretty rare I'm happy to stick around for questions too okay great thanks going