Scale By The Bay 2018: Sean Glover, Fast Data pipelines with Akka Streams and Alpakka...
Recording: Scale By The Bay 2018: Sean Glover, Fast Data pipelines with Akka Streams and Alpakka...
you so thanks for to my talk title is already just described so let's begin it's a little bit about me I'm a principal engineer at at life and I am a member of the FASTA the platform team which is a team that's building a stack of software sort of like cloud era or Hortonworks does for Big Data we do something similar for stream streaming platforms so we have a number of stream processors we support like spark and flank akka streams kafka streams and then - and then like the underlying infrastructure like caki and kubernetes or DCOs I specialize in the the katka ecosystem on this team so as a result of that I contribute to various projects within that ecosystem including katka itself so the the cactus dreams gala DSL I was a part of getting that into Apache Kafka alpaca cafe itself which was formerly known as reactive Kafka made a number of contributions there Stream Z is a chaka on kubernetes operator based implementation which I've been involved with a lot this year and DCOs Commons is just a library to help run stateful workloads on on DCOs and and it's used as part of DC west CAPTCHA to run Kafka on DC OS and Apache mezzos so alpaca is a open source initiative and it's a it's a initiative to bring it to integrate disparate systems together so it's a container project it's not really it doesn't really have its own code it has documentation maybe it'll have some code of its own in the future but it's really just bringing together a bunch of connectors to integrate various systems with akka streams to help build these stream aware reactive pipelines and in Java or Scalla so it's a very similar technology to apache camel or cap connect if you're familiar with those but it's not configuration based it's not a frame all these connectors are available as just library dependencies so you you include them in a program as as you would any other dependency in your pom.xml or build SBT so here here's kind of a logo cloud of some of the connectors that are supported I put them into three categories here the cloud services data stores and and messaging so for cloud services their support for AWS s3 sqs dynamodb pub/sub on google cloud and then mostly popular data stores that you've heard of we have connectors for those as well and of course messaging so MQTT is a new well there was an existing contribution there's a new one that looks really promising that just came out AMQP implementations like RabbitMQ JMS implementations and then of course Kafka itself so the alpaca capture connector is obviously used to connect Kafka to akka streams it was formerly known as a kiss Reims Kafka briefly for about a year and then before that it was known as a reactive CAFTA which most people probably know it as who've been using it for a while and it actually predates the alpaca community initiative by a number of years it was started by a company called software mill and it was identified early on by the Akademi a really promising project to to integrate with CAPTCHA so they broadened in in-house to the akka github organization and and either aided on it made it even better so it's used by a lot of people in the scale ecosystem it's used by a lot of light bank clients in production we provide support for it and it's used internally in a bunch of life and projects too including legume which is our micro serve service framework so it's used as a optional plugin for communicating between micro services it's also used in a commercial product called faster the pipelines which we talked about in in early next year and hopefully goes his first release and quarter one next year and it's it's it's bundled it's so it's used by a lot of people in production today and but but people might not have heard about it as much or at least not lately so I want to do this presentation to talk more about it so some of the things that akka streams brings to the table are things like backpressure which we're going to talk about in the next couple slides so stream processors today like spark and flank they do things like massive they support massive volume massive aggregations across like lots of different servers really well they have a lot of good streaming semantics but they know they don't support back pressure and that that's one of its defining attributes of active streams active streams was originally created to make streaming semantics easier within an actor system so if you're familiar with akka your your basic components are your your akka actors the messages that they send between each other asynchronously and you can obviously model any kind of communication you want between with those components but modeling complex streaming semantics can be difficult so akka streams was created to make that easier to provide like kind of a consistent implementation of these semantics and can be used internally within an actor system but it can also be brought into any old JVM project and used by itself you don't need to know anything about akka actors to start using it and of course it's using actors under the hood but that's transparent to the end-user unless you start implementing more advanced features of it so here's a basic pipeline each one of these components you need at least a source and a sink obviously a source has one outlet sink at least one outlet a sink has at least one inlet and then you have this arbitrary number of flows in between and messages obviously flow downstream from the source to the sink but you also have this other channel that's going upstream and this is for the the back pressure protocol in akka streams which we will talk about in the next slide one another unique feature of active streams is that it can support really complex graphs so most of the popular stream processors only support a cyclic graphs which means you can't have cycles or loops within your graph the backing stream is actually supports this if you have a need for that I haven't seen a lot of systems that do that but if you need to implement some sort of feedback loop where downstream flow influences an upstream flow for example you can do that if you want the active streams is is built that the back pressure where the inactive streams is implemented using this specification called reactive streams and this provides a synchronous stream processing with non blocking back pressure and it borrows a lot of the same principles you would find in the reactive manifesto to help build flexible loosely coupled scalable systems that communicate asynchronously and the spec itself was was spearheaded by Lipan but it has contributions from a lot of companies in in the streaming space that a lot of household names like Red Hat Netflix and pivotal have all contributed towards the spec and so it's active streams is a JVM implementation of it there's other implementations out there that run on the JVM and you can also implement it on another language runtime if you want believe there's a couple examples out there this is just an example of three and what this slide is calling out is that the the act the reactive stream spec was actually introduced into JDK 9 under this pigeon set of types under this package and the names aren't exactly the same as what the spec defines but they're semantically equivalent and this is this is cool because each one of these libraries implements their own implementation of reactive streams but once the community moves past Java 8 you know probably one once people start using Java 11 a lot will start seeing releases of these guys that you know can do away with their own internal implementation and use the the flow implementation in the JDK and that will reduce some boilerplate obviously and also make it easier to interoperate between these so this is a short slide demo of how backpressure works so I've been referring to these as components I'm going to use a actual name which is a graph stage within active streams so a graph stage can be a source could be a flow can be a sink and it has a lot of the same characteristics as an actor in akka it has a mailbox which is represented by these these rectangles underneath and in akka by default if you your mailbox is unconstrained in size it can be very large which can present problems you can imagine that if if a actor is not processing messages fast enough those mailboxes can eventually fill with messages and you can run out of memory and acta streams they actually constrain the size by default this is what helps enable back pressure to actually function so let's let's start with the example oh the other thing I forgot to mention is that like a cactus they also only process one message at a time and in order to achieve concurrency you basically scale out instances of these and this is really good because it it's one of the core features of akka to provide a easy way to model concurrency so common problems in concurrent systems are you know closing over code that multiple threads could access and changing state where that can introduce you know race condition things like that you don't get that in in actors because you have that guarantee you're processing one message at a time and it can mutate state within that actor as much as it likes because it knows nothing else is closing over it at the same time it's called the single threaded illusion in NACA and you get the same thing in a flow and then starting in a graph stage and then you can you can also encapsulate state within these graph stages as well so back when you start the NACA stream the most downstream components like the sink are the first that act nothing no messages are passing through until the those downstream components start requesting messages so in this case the sink says or requires messages it doesn't have anything in this mailbox that's ready to receive them so it sends this demand request upstream using that that channel I showed on an earlier slide and it sends it to its immediate upstream graph stage so the flow the flow doesn't have any messages to provide it so it sends its own message upstream to the source and then the source does what it does and loads those messages from somewhere so in this example a Kafka topic and then now we can start satisfying those demand requests that were made previously so the flow will start getting messages and processing output that it can now deliver to the sink because it knows the sink is waiting for messages to another principle of back pressure and why we're actually doing back pressure is because we don't want to we don't want to over flow graph stages downstream or systems downstream it doesn't have to be within the same JVM so to simplify how this works I just have to graph stages so in this case the flow says I could I can handle some messages now and in a naive implementation of a pull model if you might see you might think that it would send the demand requests for every message that it needs but you can imagine that that doesn't scale if you have millions of messages going through this this pipeline this graph then obviously you have a million demand requests to so that that doesn't scale very well so the compromise is this dynamic push-pull model that we have in the reactive streams so the flow knows exactly how many messages it can accommodate because it knows the size of this mailbox is bounded in size so it says in this case it's the size of five you can tune this of course and it sends a demand request with that metadata in it that it can handle up to five and this is sort of a pull side of that that model and then the source can immediately satisfy that by batch sending or pushing five messages and you you can see that you know as these requests the demand requests get larger in size that you know there's less there's less overhead going on that back channel back pressure channel which which help with performance and the whole the whole point of doing all this is that we want it we want to solve this problem the common problem in streaming is that we have a fast producer a slow consumer and this is this is a good way of solving that problem so this use case would never happen because when the flow is at capacity it would never it would it would start back pressuring and it would never receive another message from the source until until it has room and sends those demand requests so this is a simple kind of end-to-end akha streams example everything you need is here cept the imports so I can fit on the slide but you basically have your your akka actor system that you spin up because it's using actors under the hood you give it a name give a materializer to instantiate the actors and then this example is just you know very simple it's just a small iterable in memory interval going from 1 to 100 and calculating factorials and this scan is a graph stage this is a stateful graph stage sort of like a fold but instead of folding to completion which we we might not have a completion in in a streaming use case obviously it will emit its state or whatever you want after every iteration and then this is you know being formatted and put into a sync which in this case is just a file so why why would you use alpaca cackling we talked the backpressure complex event processing I'll get into later on this presentation it's a really good solution for that and then the if you need to model really complex graphs that might may have cycles in them for example it's also a good choice so here's a alpaca katka program end to end so in this case all the examples are going to be consuming in this presentation all the examples will be consuming from Kaka and in producing back to Kafka so a consume transform produce workflow and when you're bringing up alpaca Kafka into your project you're basically getting a library of sources and syncs and flows and then some other types in order for the to work with these these sources and sinks and flows so this is a committable source which is basically using offset committing within Kafka which we'll talk about shortly it's can it's sending messages downstream along with some metadata about where those messages came from with offset they are and so on a simple topic subscription where the messages are coming from really naive transformation which is actually not really doing anything except passing that message along and then a sink which is basically producing it to the the subscription found here and committing that they were they were read the messages we consumed were read and this is just some extra boilerplate to gracefully shut down that stream to stop the source drain all the outstanding messages and and wait for everything to be completed in the sink before shutting down a program so I'm going to talk about a couple Kafka features in a more a little more detail in order to better illustrate why or how you use certain features no packet katka so consumer groups is first so consumer groups are how we handle how we scale up our consumers efficiently so that they can handle more messages so to come the common use case when we don't have enough processing power on the consumer side as we start seeing this attribute of consumer lag and it can be represented in tools this is a katka offset monitor and that this blue line is representing the latest messages lives offset available in the topics were consuming or an aggregate of them anyway and then this I think is green this line represents the latest offset that we've read in that consumer group and you can see there they're growing over time and this is a bad situation to be in ideally they would they would start approaching that there would be very close to one another but when they when they grow like this if they grow indefinitely then you have a problem eventually your consumers reading later and later data and eventually kafka which is the buffer holding all these messages in between it's going to start evicting data because it's going to run out of disk eventually so you need to you need to handle this situation and we'll see how to do that in a second but what you want to see is is these lines converging and that your your lag approaches zero you know so you have very few messages between these two so this is the same thing represented as diagram so you have producers with a combined throughput of 10 megabytes a second for example and then the consumers can only process nine megabytes a second so you get this growing lag over time and then there's no back pressure so we can't tell producers to slow down and we don't want to because in a lot of like the use case for calc is to absorb a lot of volume and to you know kind of prevent that volume from cascading into our internal infrastructure that might not be able to handle it or at least that's what most a lot of people use CAC before so there's no way that tell these producers to slow down so we handle this by adding another consumer you know and then the aggregate throughput of our consumer cat now now could reach 12 megabytes a second and eventually that that that lag will will fix itself over time so what's happening when we add a consumer or remove a consumer is a called a rebalance of the consumer group so consumers or members of the consumer group I'll receive unique partitions so they're not processing the same data twice rebalances happen when you add a client remove it if the topic itself changes more partitions are added to it for example so in this case we'll just add a client the client comes along it has the same group ID and it communicates with katka consumer group chord neither so it says I want to join this group the group coordinator talks to the leader of the consumer group which is usually the oldest member of that group and it has some logic in it to determine how to reassign partitions efficiently across this group so there's some default strategies that most people use within Kafka but if you have a need to implement your own you can do it in the client which is why this exists in the client and not in traffic itself so that users have the option of extending that and easily without having to like update the cluster for example so the leader sends a new client petition assignment back to the coordinator which sends it back out to the clients but before they start consuming the haven't there's a hook where they have an opportunity to do something with the partitions that they were assigned and the petitions say they no longer have that have been revoked and and balanced to another client and a common thing they do is actually commit you know the offsets they have for those revoked partitions so the latest the messages they have kind of in flight that they're processing oftentimes you want to commit you want to finish processing them and then commit them to the offset log within caki once that's done then they then they continue reading from the consumer group with their new partition assignment and how that's done in so so in this slide that these hooks are available in the catholic client means aren't their event handlers to handle when partitions are revoked in the sign so in alpaca caca we support that with this rebalance listener which can be any act actor you don't have to create a new one just for the job it could be an existing one it could be in the same JVM and it could be a remote act actor and it just needs to handle these these two messages and then this is showing the example where we we want to commit those messages from the partitions that have been revoked so that the next client that receives that that partition can read from the latest offset we don't have duplication going on so you you get a reference to this actor and you pass it along with your topic subscription when you when you spin up your stream that's everything are you said transactions so this is this is the meat of this presentation I think a lot of people they've probably heard of consumer group balancing and all that but they not might not be familiar with transactional systems available in CAF canal so we'll spend some time talking about that so transactions naval atomic rights across multiple kapha topics and partitions it was introduced as part of Kafka's zero 11 last year June 28th and it wasn't possible before that there were situations where that perdu their corner cases where producer could send duplicate messages or send messages out of order for various reasons I a lot of time to get into so Kafka transactions is actually a number of features implemented in order to provide really good message delivery guarantees in Kafka so people familiar with messaging system should be familiar with these but the just a refresher most once is basically I want the latest data now I don't really care if there's gaps in the data I just want to process data as fast as possible a good example of this is like a game client a game server that's using the UDP protocol it doesn't have all these strict kind of acknowledgment requirements built into the protocol itself so you can model that in traffic if you have a need at least once is what I find most people settle on they want to process every message at least once and this is basically waiting for acknowledgments that messages were received by the producer before before they move on and if they don't receive those acknowledgments and they retry so the consequence is that as possible you can have duplicate messages which is why it's called at least once and then exactly once it's sort of this holy grail of message delivery guarantees that theoretically isn't isn't possible at a low level low protocol level you can you can provide those guarantees with TCP you know can do at least once UDP that's good at most once but there's no way to guarantee exactly once which is which is basically that you know messages are sent to their destination once processed only once sent in the order they're sent in the correct door then you can't provide all these guarantees because you require more context at a higher level of the network step then though a lower level can provide so the way it's done in CAPTCHA is by faking certain we're basically faking exactly once by implementing a couple features so it I like to call it at least once message delivery with effective idempotency guarantees on the sink so we'll do things like retry until we get acknowledgments from the sink we have an idempotent source so the traffic is doing D duping for us somehow and we enforce that messages are not processed more than once and and that they are sent in the right order so the way that this is implemented with within Kafka's is three main features ui impotant producer multiple partition atomic rights and consumer read isolation level so we'll go through each one of these quickly this is probably the easiest one to understand it's just we in this example we have a client sending a message along with some metadata a sequence number which is just a increasing number from zero and you have one of these sequence numbers per partition that you're producing to and then a producer ID I guess generated for the first session of this producer so it sends its message broker persisted to a partition and then it sends an acknowledgment back to the client and if this can't if this can't succeed for whatever reason then naturally the client is going to retry and when it retries it sends the same information again but the broker is expecting a new message with a incremented sequence number of one and so but it but it knows that it already persisted this one so instead of failing it just sends an acknowledgement saying to the client hey that you know we already received this message it's a duplicate but you know carry on and and give us the next message if the broker if the broker set was sent a sequence number of like two and we've only received zero then that's a that's a fatal exception and basically the stream will shut down at that point we can't recover from that we can't provide those those message order guarantees anymore because something went seriously wrong the atomic rights are used to as part of the commit process in cafe transactions so when you commit a transaction you're basically kicking off the an implementation of the two-phase commit protocol which people should be familiar with if if you've worked at distributed systems so how that's that's working in cap is that the first phase is just preparing the commit so all the messages all the data we want as part of this transactions been persisted where we want it and then the second phase is this actual atomic operation we rewrite to various partitions all at the same time one or nothing they're all or nothing kind of operation so we commit these control messages to the partitions we're producing - and these are these are messages that are transparent to the end-user they're just handled by the brokers and the clients and we'll see why they're used in a sec but they basically say you know everything before this offset has been part of a successful transaction transaction log is updated with a successful transaction message and then all of the messages we were consuming to transform and produce to another CAFTA topic are committed so it were basically saying everything we consumed is part of this transactions also being committed so this is happening all at once and this naturally flows into the last thing which is the read isolation level so you have these clients reading the output of those transactional workloads and you don't want them to process messages as as fast as possible you only want them to process messages that are part of a successful transaction so it's when you set this isolation level in your client to read committed it's basically waiting until it sees the control message before it consumes any of those previous offsets so guaranteeing you know those guarantees of the transaction are are are effective and this the this should illustrate that there's a there's a problem with transactional workloads so when they implemented this they did some benchmarking with a you know a standard Kafka application and a transactional one and even though there's a lot of overhead in this kind of system they didn't notice really much of a degradation in performance the penalty comes with the end-to-end latency so if you have multiple transactional workloads here the fact that you are now batching a series of Records is part of a transaction is what causes you know end-to-end latency within your pipeline so if you have a lot of these in serial sequentially like this the the the ones farther downstream are gonna have a longer latency just because of the fact of these batches so that that's something to watch out for when when building systems with this so the way this is implemented in alpaca cap is there is a sink that so there's a transactional source which doesn't really do much it's kind of similar to the committable source then all the magic is in the transactional sink and there's also a transactional flow and it it's basically committing all the messages as part of a transaction every interval so it's using 100 milliseconds by default and the design of this is very similar to have CAFTA streams does it itself they they use the same default actually and it commits every interval so I'm just gonna show a little slide demo how that works so when we start we begin a transaction and while the transactions open we got messages flowing through the system they're added to the mailbox they get processed successfully we have acknowledgments that these are part of the transaction sorry they are produced to the target topic topic partitions we have some other messages waiting for an acknowledgment that they've been persisted and everything's good keep receiving messages for a while and then this and then and then this loop ticks so it the commit interval lapses at 100 milliseconds elapses and instead of closing over instead of doing something immediately within ACTA the way it works is in order to give that that single threaded illusion where we're only processing one message at a time we the the timer will actually send like this meta message into our own mailbox and we append nets and met the mailbox and then once that message is processed it kicks off a bunch of things so the first thing is it starts back pressuring it stops sending demand requests because we can't add more messages to the transaction until it's committed so it does that and then we start committing the transaction so we send all the consumed offsets you know that all these messages that were can all the consumed messages that transformed into these messages that are part of our transaction are read and and then we can actually commit the transaction one thing I didn't show is well when we do a commit we're waiting for we have to drain these messages waiting for the acknowledgments we wait for all those to become part of the transaction we commit the transaction and once that's complete then you know the system starts working again we start sending demand requests we're not back pressuring anymore and things continue as they did before so here's the the code you know packet caki to do that it's almost identical to what we had before in the previous example the only difference is we have another source another sink we have a transactional ID but this demonstrates how easy it is for the end user so although the what's happening under the hood is fairly complex for an end user it's pretty easy to get started with with transactions in it it's reducing a lot of extra boilerplate you'd have to otherwise write too these guarantees like deduping logic and message ordering logic and things like that so a complex event processing is is used to you know do take an event or a message of some kind do some sort of complex analysis of it so I just have a little slide or a little little a slide here that's demonstrating how that's done in a kiss Reims we use something called the ask pattern which is basically a way to model a request and response workflow asynchronously and the ask patterns used to communicate with other ACTA actor systems and then it it asynchronously waits for a response or time set so you can use this to talk to another actor within the je within the actor system that the stream is running in or it could be a remote actor and its own system and it could also be part of a active cluster so it could be spilled over across multiple machines and you can model like any kind of complex business logic you can imagine with innaka so this shows how how easy it is to to communicate with actor systems using active streams this is the this is the code and what's happening in that previous slide so we have this actor here it's receiving some kind of message processing it by probably creating more messages sending them throughout the actor system or across a cluster it's implemented in Scala using the question mark as part of Scala DSL and one property that's important is this parallelism factor so a map async is basically doing things concurrently but maintains message order within a customs but in order to keep this efficient we want to have a parallelism factor of at least greater than one so that the actor always has like a message waiting in its mailbox that can quickly DQ in process efficiently persistent staple stages are really useful to recover state after failure create an event logged for auditing purposes or to share state probably more importantly so this is a big part of CQRS and event sourcing is a sub pattern of that if you have a lot of rights and you want to share that stateful information that those rights are creating in a efficient way then you'll use this pattern event sourcing most likely lots of database is implemented Kefka streams k tables implemented so here's here's a really kind of convoluted diagram how that might work where you get these requests in commands or queries are handled by a request handler which emits an event and then that event gets handled by the persistence plugin for akka which puts it into an event log which is a append-only event log it also gets handled by the event handler which is maintaining the state or enriching the event with state stateful information and that that's how it works what in normal operation and then the recovery phase is like if it's recovering from failure we don't consume any messages until we replay this full event log through this workflow again which reconstitutes the state and brings it back to what we want so unfortunately this isn't a concrete implementation yet in active streams there is a proposal for doing it by martin krauser i'm hoping there's an enterprising individual in this crowd that might pick it up because it would be a really cool contribution to make to akka streams and it would be fun to work on I think as well so that's it I'm gonna close with this release notes from the latest I'll packet after release it was released as of November 6 a couple you a couple weeks ago it's using the new asynchronous client API so it's probably the most notable change but I want to also demonstrate that it's an active project lots of user contributions has its own team at light Bend as well so it gets a lot of attention and I'll conclude with saying that I hope that I've done a good job in showing different parts of the akka toolkit it's just one tool as part of it's just one tool in the tool belt all these tools are kind of lying on the floor ready for you to pick them up and use them and I hope you consider using them to build your next your next streaming platform so thanks and I'll take some questions if we have some time no time okay thank you [Applause]