Reactive Stream Processing with Kafka-rx

BDSBTB 2015: Thomas Omans, Reactive Stream Processing with Kafka-Rx

Recording: BDSBTB 2015: Thomas Omans, Reactive Stream Processing with Kafka-Rx

awesome thank you everyone very very happy to be speaking here before we get started could we all just kind of maybe stand up stretch I know it's been a long day and it's good to get the blood flowing maybe one of those weird ones where you use your brain I know we've probably done a little too much of that today but maybe we can keep it going so yeah my name is Thomas Homans I am an engineering lead I go by X be at the internet and I blog sometimes it's same sake I work with a company called Commission Junction were in the affiliate ad space we're a pretty large company but we're a very old company and I'll talk a little bit more about that later but first it's just great to meet everyone and again thank you for coming so why am I here I want to talk to you guys about a library called Kafka RX and what this library is is just a high-level wrapper around the producers and consumers that ship out of the box with Kafka here's an example of some of the code that you might write with a system like this so here we have a tweet stream presumably living in Kafka and at this state we're assuming it's kind of raw data it's just just a stream of tweets however it got there is not really the concern of this program but what we do want to do is we want to look at it and we want to group it by this hashtag function that's going to look at the tweet and extract some information about it so this is really cool because you guys have probably seen these group by on these flat map functions in Scala collections if you've been doing that style of programming but here we're doing it over this you know asynchronous event stream so when we when we group them by their hashtag now we have sub streams and we're able to process those sub streams independently and still in the safe synchronous context something that we found was really cool is that if we could if we could get a stream of producer records these things that come with Kafka that you say this is how you get a message into the system then now we can ring then into Kafka if you just handed a producer we used implicit to get this this behavior and we just attached it to a to the API of the observable also we wanted very specific commit semantics and so here we're actually calling message commit but it might be a little weird because it's actually coming from the result of saving to Kafka and this commit is committing back in the original source stream of tweets so we've had a whole transformation that's been going on but we're keeping track of where we were this is very similar to storm check pointed so why did we were right Kafka RX there's a lot of stuff out there and I'm sure many of the people here are familiar with the kafka story are familiar with the data integration problem the legacy systems problem and as it turns out cafe is an amazing way to separate to separate your systems from one another you can use it to rather than so rather than two discrete components of your system knowing the intimate details of one another unfortunately they know the intimate details of Kafka instead this is a trade-off we think is is worth it but it is absolutely a trade-off so our X is a library for stream transformations and Kafka is a system for streams so to us this seemed like a natural mapping the existing clients didn't support the sort of things we wanted to do we wanted to peek into the stream and the coffee client just didn't support that it could but it just wasn't wasn't what people were using it for before and so because of this you know we were here today CJ is like I mentioned a very old company and that means we experienced a lot of time some things grow with time like data for example accumulates with time the longer that you're running a data system the more data you're going to come in that's more going to the more data you're going to be able to process unfortunately the longer your system is around you're going to get this other really strange thing called technical debt and what this means is that you make architectural or infrastructural decisions and they start to crystallize into your applications and it makes it very very hot to change your applications and it's very strange because these two things are completely at odds on one sense data loves to change data loves to evolve as new requirements are added or removed we're continuously changing our schemas we use Avro for this and debt on the other hand makes it impossible to do exactly those sort of changes that you need to do if you want to continue moving so how do we reconcile these things how do we if we know that our problems are changing and they're changing faster than our tools so how can we keep up I think it's pretty pretty simple all you have to do is find the simplest solution that solves your problem and then use it rich Hickey stated this very eloquently in a talk called simple made easy if you haven't seen it i recommend you go watch it as soon as possible but the general idea is that you should battle complexity you should not accommodate it you shouldn't you shouldn't have very complex systems but have ways of dealing with that complexity instead we you should try to do is you should we'd it out at the root so what sort of stuff of them I going to be talking about today primarily one thing streams stream stream stream stream streams in order to talk about that I think we're going to have to go over some terms and definitions because I think stream processing is one of those things that means a lot of different things to a lot of different people and I just want to make sure that I clearly let you guys know what I mean when I say these words after that we'll get to a little more theoretical stuff how what the problems in stream processing systems are and finally we'll get to the practical usage so what is stream processing i defined stream processing as a continuous directed transformation and I think all of those words are very important and you might say that those words compose into a larger into a larger phrase so first the first word is continuous what does that mean it means that it's continuing it means that if there's this question in the back of your head of is this thing still going on the answer is always yes it's also directed and that's directed in the causality sense so we say that a implies B here in these states and that is the direction it's not that B implies a that's bi-directional so we're talking about things that go in one direction generally the direction of time and we're talking about the way to transform them so here's a slide you guys have probably seen before but we need to talk about it if we're going to talk about distributed systems so what are concurrent streams if we're talking about this sort of consistency where one thing happens after another and we could say that the first state implies the next state well what happens if two people both one record 12 someone's going to have to decide if it's very important that you only give out that record 12 once I if they if they both come at the same time you're going to have to either make one of them wait or give out a duplicate neither of those are good options so what can you do if this were like a checkout line and there were a bunch of people you just say we'll add a new check out so that's exactly what Kafka does with its partitions of its topics so here we have rights going in concurrently and there's no problem because they're handled in parallel what's really interesting about this and I think hits concurrency on the head is that we can now no longer say whether records three happened in partition zero one or two first there's absolutely no information here that could tell us and this is what it means to be concurrent it's two different dimensions of time now you can always rebuild it back right you could timestamp your data and you can reassemble it but the concurrency means that there is no logical ordering oh by the way we've been talking about Kafka I'm sure you guys already caught this but we've been talking about Kafka so Kafka says it's a high-throughput log messaging system I've built for publish and subscribe patterns but honestly I think of Kafka more is just a system for distributing concurrent streams the other part that I want to talk about is reactive programming reactive programming is one of these things that's probably said more by business people than programmers and when the programmers are talking about it they're usually not talking to each other about the same thing but I think there's a really simple definition of what this is and it's just a matter of push verse pole in our standard systems we're pulling data out of them we say hello do you have any data for me and then if they say they do we extract it from them now contrast this to call back based systems where they push data into you what you do is you say here's what I want you to do if you if anything should ever happen in your system and then when those things happen they call your call back and in this sense you're forced to react to their events you don't get the opportunity to not call your call back now maybe you want that pole model in some cases specifically for that reason it's an inversion of control problem the other thing that goes hand in hand with stream processing is functional programming and I also define functional programming pretty simply you're doing functional programming if you're programming with functions that's it you don't need to do anything fancy you just need to program with functions now it's a gradient right the more you the more functions you're using and the strange the stranger the ways that you use them become the deeper down that functional labyrinth you're probably going but in general just program with functions so we want to have value expressions right we want it we want any expression to be able to be replaced at will with the value that that expression evaluates to and vice versa in order to do this you need something called referential transparency and in order to do that you need immutable data usually also there's this notion of sequence operations this is the things like map and filter the things that take what we call higher-order functions because their functions that are defined in terms of functions now that that thing right there being defined by something that is like yourself we call that composition so if I'm defined in terms of other functions then I am composed out of functions so what about reactive x this is one of the libraries that I really want to talk about today who here is familiar or who has used reactive x cool so almost no one excellent so reactive extensions came out of Microsoft a few years back and they're basically a library for working with streams and specifically for transforming those streams they describe themselves as a library for composing asynchronous and event based programs now remember what I was saying about composition how how you can define something about yourself based on other inputs that are like you this is a very interesting property and we'll get into it a little bit later first let's look at the foundation of reactive extensions the observable stream so the observable stream is the observable stream is just equivalent to an iterable so with an iterable you say oh can I get an iterator and when you have that iterator you can do that data extraction thing you can say okay I'm ready for more data I'm ready for more data now contrast that with the observable where instead of getting an iterator you say subscribe and when you subscribe you have to project your call backs into that thing you say here's what it means for me to handle a new event here's what it means for me for me to handle the completion and also we have to deal with errors because of the inversion of control but there's some interesting mathematical properties if you are that this last talked about category theory you might say that these two things are duels but the gist of it at the end of the day all it really means is that you have these sort of operations that you're used to using like filter like select likes can like reduce and you have them available on these data structures that span time so fold is something that you they should not be able to be done on unbounded lists and the reason is because fold is this sort of pairwise operation that takes two values and flattens them together all the way down to the end of the list so what can we do instead if we don't know if the end of the list is ever going to come what we can do is we can say okay one goes and that merges into one and then one and the next state is three and then the former state of three and the new state merges into six and six and four merges into 10 now this idea of merging old state and new state is going to be very very powerful and we'll look at how it can help us a little bit later another and I think maybe the most powerful idea from reactive extensions is this idea of streams is signals if you think of yourself driving down the street and you think of you think of yourself has a stream processing system there's all these streams of information other drivers on the road what the weather is like today and you're continuously they're processing it now remember you're also traveling through the direction of time so you fit that model of a stream so what happens when you see a light change there's this new thing of information that you haven't been thinking about it wasn't in the forefront of your mind and yet when it comes in you're changing your behavior because you were implicitly defined by this other stream now how does this map into our actual programs you might think about cancelling a request of an ajax user right so so you've got a very complicated pipeline that you just triggered and it's probably going to take a few days to run and then suddenly the users shows that they're not interested anymore they say oh well that's not very interesting and so rather than letting that thing happen you can actually intersect that stream with this pipeline and now you can just completely collapse it and not do any more work so what you're doing here is you're composing streams and this idea is incredibly powerful this is the main idea we wanted to use reactive extensions some of you are also lucky enough to have seen Martin clever men talk about streams in the UNIX philosophy Jade touched on this earlier as well today the general idea of the UNIX philosophy is that you should do one thing and you should do it well the single responsibility principle you need to pipe your inputs to your outputs but it's not just about this pipe operator that's not where the interesting thing is happening the interesting thing happens because every program that's in this system programs against a shared abstraction they all agree that there's this thing called a line and it's any number of bytes delimited by a newline character and that if you see that that shall be your input and that shall be your output and because you have this shared abstraction that everyone agrees on now you can compose your tools together but it doesn't happen for free if there wasn't that line you want to be able to compose so Kafka becomes like this pipe but it's still your responsibility to make the shared abstraction in your own domain so Kafka is a distributed system I said it was for distributing concurrent streams so how do we think with distributed streams there's a whole sort of question that a whole range of questions you have to ask yourself when you're working in these systems because of the nature of the network so you might have to ask did my message get delivered so the reality we live in is that networks fail their asynchronous and delivery is not guaranteed so i might say hello take my message and then you say all right i'll take that message and then the next time it comes along I say take my message and then someone cuts my cord I have a brain aneurysm and I fall down now when I come back I've got two choices here right I can either say well I don't know that it was guarantee that my nut message which was was received or I can say oh well it was probably received and you know it's not a big deal anyhow this is starting to define some delivery semantics the first one where you try to deliver it again until you know for sure that it has been received we call at least one semantics because we're going to try until it's there at least once the other one where you are just really optimistic and say yeah well it probably worked that's called atmel most once you sort of start at zero and work your way towards one the other thing that you might have to ask is have I seen this message before so if you chose at least once because you knew you needed reliable delivery now you have to deal with duplicates so you have to say have I seen this message before the other thing that you have to deal with is message reordering so if you are in three different states and you're all simultaneously trying to send an event well the geographic distance alone is going to make those happen in a different order or be received in a different order n geographic location isn't the only thing that's going to cause data scheme you can also have internal agency you have GC pauses so you have to ask yourself did are these the order that I intend them to be in finally you might ask have I seen this before so how do we do how do we deal with these problems these problems are very very challenging and not something that we usually deal with as regular application developers but they're very old problems and they've been around forever because they're so fundamental the first tool that we have and the one that we're probably the most familiar with is this causal consistency this idea that one thing happens before another and again this is directionality we say that it was in state 1 and then after that it was in state B there's no in-between about this there's no ambiguity right it happens and and that's the end of the story but it's not just enough that there's causal consistency that you know that this thing implies this other thing the other thing that you need is atomic broadcast if you're in a distributed environment it's important that everyone sees that update at the same time you can get this for example by using zookeeper I recommend you do the problem with causal consistency and sequential consistency is that it's the most expensive of all so what happens if you need this sort of reliability but you can't afford to pay the synchronization costs well there's also another tool out there available to us programmers called eventual consistency and I think when eventual consistency first came out people thought about it as maybe consistency probably not really consistency or kind of but actually some very some very smart people Mark Shapiro and friends have demonstrated that there's this interesting property you can get where so long as everyone receives the same set of operations everyone will be in this the same result in n state so here we have three nodes we have nodes x1 x2 and x3 and they're doing two concurrent operations we don't know which happened first but again they're concurrent so it doesn't matter what does matter is that those operations propagate out to the other servers and in fact every operation must propagate out to every other server now there are a lot of ways you can do this you can use gossip protocols but it's generally a solved problem now if you were paying attention to that last slide you might have seen that all of the magic was happening inside of this merge function that there's this interesting property and these things are called crdt s which is commutative or convergent replicated data types and these things are made specifically for a distributed environment specifically to be run in parallel and all you need to do is you need to be able to take two states states a and B and you need to be able to merge them into new state see it's important that when you do that merge operation that it is bigger than states a and B in the causality sense is bigger right that that it is the new state of the system and not a former state of the system because that would violate the principles and also because of the way the network works because you might receive these messages more than once because you might receive them out of order you also need to guarantee a few mathematical properties you need to ensure that your merge function is can putative and what that means is that I can take this new and this old and I can swap them around and it won't change the behavior of the function there's also this property called associativity and that means that I can flip these applications to merge and it's no different than had I not done that finally it needs to be idempotent which means that if I call the thing once versus calling it many times there is no difference an example of an item potent operation is like setting X 21 no matter how many times you say set X 21 the end result is going to be 1 in that database so hopefully you can see why these mathematical properties actually link directly back to the nature of the distributed networks that we're working on now that we kind of understand the way that all of these things work we can start to talk about the different models the different techniques that people are using the most popular are the continuous operator models the one record at a time style and the micro batch models the many records at a time style a lot of people come in from functional programming are going to prefer the micro batch because list operations are their bread and butter and anything that can make it more listy is something that they're going to go after the other people really like one record at a time because it's like these little propagator Network's it's like these little nodes in a graph and you can control them independently and discreetly you'll have some other options do you want to use local state and just have everyone have a local cache of the data or do you want to distribute out your state again these are the decisions that you're going to have to make based on performance based on what makes sense for how you're going to use it when you run your stream processing job is it going to be deterministic is it going to do exactly the same thing every single time you run it or can you afford non determinism can it be observably deterministic is your data set bounded something that's known to end like a web session or is it unbounded is it something that's never going to end like the procession of the seasons and do you need to when you're like merging your into your result set does that need to be synchronous can no one else do that at the same time or can you afford it to be uncoordinated so you might be kind of confused here saying well this is a lot of stuff this is kind of like a paradox of choice and which way should I use honestly there's no silver bullet and like everything else I'm not going to be able to give you one answer but use your head and look at the problems and see what guarantees you need to be able to apply often you'll need some combination of both often you'll need strong consistency and eventual consistency often you'll need local cache state and distributed state what I can tell you is regardless of what you choose your solutions need to compose you need to be able to take one solution and put it into another and bad should stack all the way up and finally different systems have different requirements so if I can't tell you what model is right for you without talking more what can we do instead I think if you form values and you use them to guide your decisions that you can generally wind up in a pretty good place so what I'm going to say is you should look for things that let you program against value transformations things that let you push the integration to the edges so if I'm going to make a very complicated system I don't want I don't want the internals of some dependent system leaking into the implementation I want to program against simple abstractions and I want to make sure that that I treat things separately again this is just applying the UNIX principles so finally what is this library look like what does it do so here's a simple consumer and what we do here is we connect to zookeeper much like the high level Kafka consumer does and then we say give me a message stream for a particular topic of course we support the whitelist wild cards and everything we're always just delegating to Kafka underneath it's very important that we don't try to go in a different direction but instead we just become a faithful model on top of Kafka itself here we tumble five seconds so we split it into five second windows so we just here in process decided I would prefer micro batch for this problem now I could have done something else I could have done map and then I would have been a little bit closer to the continuous operator model and then finally I'm able to just process it whatever makes sense for me I also said that it's important to program against value transformations and the reason is that if your systems are going to live a long time they're going to change a lot and you need to be able to define things that won't change you need to be able to say that no matter what happens in this system it should always be true that when I poke it this way this thing happens so here because we've pushed our integration to the edges because we don't even take it we program our consumers to where they don't even know anything about Kafka what they know about is a stream of values and so then we faked out that stream of values in process Kafka's producers are behind an interface so we can fake that out in process as well so we completely have arbitrary transformations and it's all at the level of data so the important thing I want to get across is that if you're around for a long long time the systems that you're working with are going to come and go it doesn't matter where your data is loaded from it doesn't matter if it's coming in over batch if it's streamed in from mobile sensors if it's typed in by monkeys at a computer the only thing that matters is how you're going to process it it doesn't matter where it's stored the only thing that matters is the abstraction you're going to use when you talk about it so how about distribution we've been talking about distributed systems we've been talking about concurrency how does this thing distribute and we also kind of sidestepped this problem we said well Kafka naturally distributes horizontally it's distributed by default and so does this thing run on Mase O's does it run on yarn docker or AWS or your new cloud of choice the answer is that that's an application concern and we're not going to be able to map that faithfully for you you're going to have to make the decision of how you want to scale your consumers when that time comes our only job is to step back and make sure that you're able to make that decision so you can run on one server or a hundred and in fact if you're programming these things right it shouldn't matter if you only have ten events a day or a hundred billion events per day you should be able to decide the boundaries of your program so you might say oh well that's all well and good but I just need a framework that gives me exactly once processing I don't care how it does it I just need it and to that all say well do you really maybe we should have a conversation let's talk about what exactly once means for you I can tell you all kinds of reasons why you probably won't want it but rather than doing that I'm going to talk about how we enable it so here we have a stream and honestly it doesn't matter what this stream is all we're doing is we're processing it as a bucket and then we're committing and that commits here takes a merge call back and what that merge call back is going to do is take the offsets that are that are global the ones that are distributed and shared and the ones that you're proposing and then it's going to give them to your to your program and give you the opportunity to reconcile so what might happen is there might be work that was unaccounted for someone might have might have done that thing where they process the data and then crashed before they were able to acknowledge it now this is known you know it's at least one so we're going to process but maybe there's some weird side effect in our processing that we really can't accommodate maybe we actually need to detect this sort of thing happening and we need to unwind or like unprocessed data now this isn't idempotent by you know in the mathematical sense but it is observably idempotent this commit also happens in the context of a zookeeper lock so this commit is synchronous across the consumer group so there's this thing here and it's very very complex and needs to be used to have a very specific way but is very powerful what you should do first is you should just accept item potency just just learn to live and love it but if you can't there are other ways around it we can use the things we learn from eventual consistency we can apply those operations to to what it means to add our thing into our result set all right so let's try to do a little demo always always the best part so that that a tweet code that i was showing at the beginning was actually real code more or less so here we're using the Twitter forge a library and of course it just gives you call backs when things happen but you can you can query tweet streams and so what we do is we we transform that callback api into a streaming API and then when we have this observable we're able to to group it by the topic that it needs to belong in when we found that we're just going to take these sub streams and put them into producer records and win their producer records then we can call this save to Kafka method when that happens we're going to map the results of this message being saved into Kafka into a log message and here we can take our producer stream in process and our consumer stream in process and just use the for comprehension in Scala to get one stream out of it and that's here producer dot merge consumer so what happens if we if we run something like this 222222 so again the maybe i'm lucky i think i might be having some cool network issues with my Kafka cluster crying oh yeah starting Kafka is always a good idea cool and now that it's up we can see we can see these log messages streaming out so what's happened is we just mapped a simple callback API into an observable stream and I'll show you guys how that's done so this Twitter for Jay library just is pretty simpler simple you have to implement this status listener like most callbacks order classes and you're given an on next handler well here we're just using a published subject which is something kind of like a promise in our X and you just get to push events into it and then the other things can read the events out and what's really nice is that when we read messages when we read messages out of Kafka we we kind of Marshall them into this committable context of a message and then when we save them in we're able to use that exact same message because there's this abstraction that's just a message that is persisted inside of Kafka and it doesn't matter whether it comes from the producer or the consumer it really is just a thing with offsets in a location in Kafka so here we're demonstrating being able to use your own form of concurrency this could easily be actors it could easily be any other method that makes sense in your system but uh yeah so so that's kind of the the demo part so why do I think that this stuff is important why do I think this is all a really good idea I think some of them some of the highest things that you can aim for is this idea of natural design I think that the best designs are have already been invented by nature they've been hardened over millions and millions of years I think that when object-oriented programming came out it kind of touched on something like that it was it mapped into our mental model maybe it didn't map into reality but it was something that was familiar to us now I think these these streams however are much more natural and much more familiar to our everyday operation we might not be thinking of it like this we might not be thinking of these transformations so concretely but I really do think that streams are a natural model to our world so Donald a Norman said a good designs fit our needs so well that the design is invisible I think when it's that when the best thing happens you don't even know if anything happened at all an example is a doorknob it's so natural to use that it doesn't feel designed you just sit there and you twist it that's how programming should be the other important idea is that streams are just lists Plus this notion of time the abstractions need to compose on to one another and here i think that they do you have this abstraction of a list and you have this abstraction of time and then you can try to plus them together to make a new abstraction I think that Kafka is a beautiful abstraction for distributing streams and I think that reactive extensions is a marvelous marvelous abstraction for transforming those streams I think we need to change how our programs work so we're pushing more unbounded lists in and out finally I think that this is really important because our tools are going to shape how we think if we're working in a dsl and that dsl is sadistic we're going to start thinking that way if we're thinking about state mutations we're going to start thinking that way so we don't need tools to make things easy for us we need tools to help us think clearly and finally we need to protect our ability to change we need to not be locked into our solutions we need to be programming against abstractions so there's this wonderful quote by Henrik Berg from Spotify and he says system architecture should support the addition of any feature at any time this might sound kind of weird because why would you want to for example add mp3s in the middle of your data pipeline well the answer is the problems change in ways that you cannot predict and so you have to be willing to accommodate this style of thinking finally what do we plan for the future for this library well with the changes in Kafka and the shift from Scala to Java we're going to port this thing to Java we don't plan on changing our consumers we still want to use Scala but we're going to make that a thin layer over the Java underbelly there's a little work to do as far as API clean up goes there's some types that we think are kind of incidental that we'd like to erase and we would like an out-of-the-box me so sore yarn layer for the consumers I think what would be best of all is if we could see these things like RX and and curator inside of kafka core instead of something that only the community is doing on the side with that that's all I got so thank you everyone and if anyone has any questions I'd be glad to answer them you