Jay Kreps - Apache Kafka and the Rise of The Stream Data Platform

Welcome.

Welcome.

There's a few seats up here. But so great to see so many of you here today. Thank you all for coming. This is Code as Craft, if you didn't know where you were. We have a regular series of speaker events, but we also have a Twitter account that you can follow to find out about these kind of events. Code as Craft on Twitter. Also a blog, codeascraft.com, that also has details about these sort of speaker events as well as things that the engineering team at Etsy will publish from time to time. I'm excited about tonight's talk. We do— we have a lot of data at Etsy. It's a big challenge to make it all work. Our guest speaker tonight has done a lot of thinking about that. I was introduced to him from a blog post about The Log, which is an epic read if any of you— I recommend it all to all of you. But anyway, he'll tell you more about himself and what he's here to talk about tonight. Please welcome Jay Kreps.

All right, thanks. Actually, can I use that mic? Does that work?

You absolutely may.

OK. Turn this on. All right. Thanks for having me. I'm really, really excited to be at Etsy, because I love Etsy. All right, so this talk is going to be about Apache Kafka, which is a system that I've been working on for about 5 years. And I'm gonna talk a little bit about why we made it and what it's for. So it's kind of like an origin story, so maybe kind of like, you know, Batman Begins, but for distributed commit log. Which is maybe not quite as good as Batman Begins, but— so a lot of this work was done at LinkedIn, and so I'm gonna tell kind of the story of, you know, how we came up with this and what problems we were trying to solve and, you know, where it's gone a little bit since then. So we got started in 2009, and so the original problem we were trying to solve was really getting all our data in Hadoop, and we thought, well, this should be pretty easy, right? We'll just take the data and we'll put it in Hadoop, and then we'll have a lot of hard work to do building fancy modeling algorithms and building products on top of it, and it'll be really interesting. And in particular, I had come from kind of like an analytics background, so I was really looking forward to that. And it turned out to be a little harder than we thought. And so we spent a lot of time on the kind of let's get the data and let's get it in Hadoop. And we also spent a lot of time on what we'll do with it. And we originally didn't really have much of a solution to this problem. So our initial approach was really kind of just like gut it out. Like if there's 12 data sources and 12 data formats, then you need 12 people one for each thing, and if we make 12 more, then we need 12 more people. And this is actually not uncommon. I think at LinkedIn at one point when we were doing resourcing assignment, like how many new engineers do you need, there was a column. So every project you would have to add some data warehouse engineers to come do stuff. And it's not ideal. You don't want that. But worse than that was the actual coverage of data we had was not very good. So at some point we kind of made a list of things we thought happened in the company, and we made a list of things that we actually had somewhere, either in Hadoop or in the data warehouse, and neither of those, you know, the coverage in that list was very poor. So we thought we had about 14% of the total set of things that happened, which is not great. And even that 14% wasn't perfectly reliable. And so I'd come from an infrastructure background, I kind of started to get interested in this problem, and I felt it was actually a deeper problem than originally I'd appreciated. We had these different data sources. They were each different. We had data that was in databases, and that was kind of one way, and we had log files, and we had metrics, and we had other things like messaging systems here and there. We had a bunch of different data sources, and all of these were kind of undergoing change. If you have a business and it's evolving and you employ engineers, they're kind of making new data all the time, and that data is changing and kind of growing and evolving. And somehow you want to be able to reflect all that data and all that change in different places, like in analytics systems, in real-time applications, in different areas. And we weren't really sure how to do it, but I was pretty sure that I didn't want to have a big team of people that all they did was parse different types of log file formats and load data into Hadoop, and so I felt like, okay, there must be some better way to attack this problem, and so I started kind of like doing almost like a catalog of what we were doing inside the company and what other people were doing in other companies and how did it all work, and I kind of got out of the kind of pure infrastructure analytics area and started thinking about kind of the flow of data and how does it get from place to place, and it turns out that that, isn't something that people have thought about that much. Mostly in the infrastructure space, people think about data systems as a place where you put the data, and how it gets in there, you don't really think about, right? And so data systems tend to focus on places for data to go to kind of rest, not data that's actually kind of moving. And so my catalog went through a bunch of things we were doing, and they were all kind of gross. So this part, you may want to cover your eyes. But we had a lot of them. So we had databases, and database data was, I mean, databases are kind of where you put your data, right? And this flowed in different ways. So we would have changes coming out of databases that would go into a data warehouse through kind of a traditional ETL path. We had the crazy stuff my team had done for Hadoop. We had a whole system of polling and propagating changes to search indexes, to other real-time systems, to populate caches. Each of these was a little different, and each of them kind of broke in different ways, right? So these were really kind of batch data flows, so data would flow periodically, like once a day. We would dump out all the data, and we would kind of load it up. These were very real-time, but they had problems if they ever kind of fell behind, and they didn't really scale that well. And then we had a totally different way for other databases. We had key-value stores, and how that stuff got in and out of Hadoop was totally different. And so it was a little bit fragmented. And that was just the stuff in databases, and we didn't have that many different kinds of databases. Then we had user events where we would kind of log out what things were happening on the website. And this would kind of get put on some shared MS thing and rsynced around and then loaded in different places. And you can usually kind of tell whenever you see rsync in some kind of critical process. You kind of know there's gonna be some problems. And sure enough, there were some problems. And one of the problems was, really it was just this system that had good access, and then we kind of made this hack to get it over here, but this data was kind of seen as log files, like not that important, but if you think about it, it's actually what is happening in the company right now, which is pretty important, right? Other things might want to know about that. And so I thought, well, you know, like, there must be some kind of, like, real-time uses of data. And there were some, right? You know, our monitoring system was pretty real-time, but it would just get stats to monitoring, and that was kind of all it did. We had some Splunk stuff for, you know, log files. And then we had this smattering of messaging systems here and there. We were using ActiveMQ at that time. Each of these systems was kind of different. The ActiveMQ didn't really have a good distributed story, but it was kind of real-time, but you couldn't hook up any of the offline things to it. The file dumps were actually pretty high throughput. If you've ever done file dumps and file copies, they move pretty quickly, but they're real slow. Each of these things kind of gave different guarantees. Some would mostly get the data there, others would always get the data there, except sometimes when they didn't, and they kind of failed in different ways. And worst of all, when you kind of put them all together, it was a big mess. So you kind of put all these different things together and you really have like the combination of every different way of doing things. And if you try and scale your dataset, you're actually scaling it across all of these different pipelines, right? So each one of them has to grow. And then if you take this and you try and make it multi-data center, then you have to like have a story for each of these different data pipelines. And so this definitely seemed like a problem. And it was a problem I was directly interested in solving because at least one part of it was my team's problem. And because I had kind of an infrastructure background and I was running part of an infrastructure team, I really wanted to find kind of an infrastructure solution. Infrastructure being the thing where you kind of work really hard and you build some hammer and then you use the hammer to build lots of things. And so I wanted something where instead of having a large team of people that were doing kind of really low-end work, I wanted to have a small team of people doing really high-end work that would somehow address this better. And this doesn't work for everything, so it wasn't clear that we would be able to really make this happen. And so we had this idea that, hey, all these different types of data, if you kind of squint and you like, you know, get back far enough, they kind of all look the same. So in some sense, they all kind of look like a stream of, you know, events or things that happened, right? So in a database, you have changes coming into the database, and it's like, hey, now the value of this row is this, now the value of this row is this, and you can think of that as a stream. Log files, you know, we think of them as files, but in some sense, the file is like incidental. There's really a stream of, you know, things that happened in that log. Metrics data about, you know, how your servers are performing. There's also some kind of regular series of updates of like, hey, these are the stats I got from the server. All these kind of things had some kind of common pattern. And we thought, well, if we could make some kind of central platform for stream data, we could really use that. And we were a little bit inspired by Hadoop because we were also working there. And we thought, look, we have this great offline data platform where we're kind of dumping everything. If we could have a similar thing that was a stream data platform where we could kind of represent everything that was happening as a stream kind of with very low latency, and provide that to all these different systems that need it, that would actually be a really valuable thing. And so we thought, well, how hard could it be? We'll try and make this. And, you know, we thought, like, we had kind of a database background in distributed systems. We thought, well, we could probably piece something together. There's a whole area of messaging systems that we don't really know anything about. So we'll get some of these, and we'll try to make something using them. You know, maybe we'll— kind of like the sharded MySQL of real-time messaging. We'll kind of like strap a little layer on top, and voilà, we'll have a distributed stream data platform. It'll be amazing. And so we thought, eh, it should just take a few weeks. And so we tried this, and we got as far as kind of getting a hold of a production data stream and really trying to see how this would work. And it turns out it didn't work that great. There was actually a lot of problems. So when you looked at these messaging systems, which was a whole area I hadn't really been familiar with, It turns out they're not really that well suited to actually almost any of the problems that we were trying to attack, except for the ones where we were already using them, and even there it was debatable. And so they had a bunch of issues. So first, you know, the first issue was throughput. Like, if you want to take, you know, event data for a consumer-scale website, there's like a lot of it, and you have to be able to process it efficiently, or it's not gonna be cost-effective to keep it at all, right? And so we did some back-of-the-envelope, and I think with ActiveMQ at that time, we would've needed, as many ActiveMQ servers as we had servers, which is like, it's not gonna be a good way to like maintain your log data, you know, by the way. The other problem was this didn't really work for batch systems, for analytics, for Hadoop, for things that kind of maybe pick up their data less frequently or that go down for maintenance for a couple hours while you do some upgrade or, you know, whatever the case may be. And it didn't work because the messaging systems weren't good at storing data. You know, they worked really well as long as the queue was just, like, in memory and almost empty, and then they kind of, like, fell to the ground and died as soon as the queue got a little bit bigger. And so that was a big problem because, you know, at least one of my big goals was, like, get it into Hadoop, and, you know, at that time, you know, Hadoop would go down for many hours when you updated it. Okay, so then we had some kind of, like, bigger goals, the biggest of which was to be able to do real-time processing on top of this data that was more sophisticated than just looking at individual messages. And that's what's called stream processing, and I'll talk a little bit more about it in the rest of the talk. And the last problems we had around the messaging systems was just they didn't really provide good guarantees. So if you want to take important streams of data and you want to get that to be the source for your search index or whatever, Your search index has to end up with the same stuff that your source system had. You can't have it skew. People will notice and complain. And the same is true for stuff related to ads or any kind of really important business area. People will notice. There's some kind of analyticsy things where you can kind of say, well, we got most of the stuff, and directionally the chart's right. But mostly people want to get their data not lost, to get it from place to place. And the guarantees there were not great. And in particular, for database changes, the ordering guarantees in messaging systems are not great. So if you want to have an ordered stream of updates and you want to do that especially across a cluster of machines, you can't really do that very well. And that's very important if you're tracking changes in a database, right? So if you update your user account and then you update it again, it's really important that the first thing go into the search index and get replaced by the second thing, not the second thing go in and get replaced by the first thing, because otherwise your search index is wrong. So those were our, you know, problems. And, you know, having, you know, being part of an infrastructure team, well, how hard could it be? We'll just build something. And so we said, well, you know, we should build something, and it will probably take about, mm, 3 months. And we've been working on it ever since then. So if anybody ever tells you that something will take 3 months, I think the moral of the story is 3 months is really the longest that any software engineer can imagine something taking. And so really what they're saying is, like, max int. That's like— that's what that time estimate means. And— but we did actually get something into production relatively quickly. It just turned out there was more to the problem than we had originally understood. And so this was really, you know, our second attempt, and this is what we spent, you know, the next, whatever, 4 years working on. So I'll say a little bit about Kafka because maybe not everybody knows about it. So it is kind of like a messaging system, so like those other systems I just said bad things about, in that you have a central cluster, these processes can connect to it and they can send a stream of messages, and different consumers can kind of hook into that cluster and receive that stream and do stuff with it. And so at a high level, it's kind of like just a distributed big data messaging system. But in terms of how it works, it's actually built really differently. And so the core idea we had was this kind of commit log abstraction or idea. And not everybody has heard of a commit log, so I'll kind of say what it is. It's really kind of the simplest possible data structure for representing change of, well, anything, but like data. And so this is my picture of a commit log. It's not very artistic, but It's a sequence of records, one after another. And it kind of represents what occurred in time. So the older things are on the left, the newer things are on the right. Time kind of goes this way. And so that could be a sequence of, you know, log entries, like this happened, this happened, this happened, this happened. It could be a sequence of database updates, like we set this row to this, we set this row to this, we set this row to this. And it turns out this data structure is kind of more powerful than you would think. It kind of looks pretty simple, and it is pretty simple, but you can actually use it for a lot of things. And, you know, the first thing that's like maybe a little different from a log file that I should explain is like the data in these entries can be structured. It could be actually anything you want, but it doesn't just have to be like free text. So people get a little confused about that, but like an Apache log is kind of free-ish text, but this can be any kind of structured data. And then also this is a little different in that it's built to support readers. So readers can read the log. And then finally, it's different in that the rows in the log are numbered, right? And this number in Kafka terms would be called the offset, and it marks kind of your position in the log. So if you're at record 6, you're kind of at offset 6. And this turns out to be actually a pretty good basis for a publish-subscribe messaging system. So if you have some source system, it's appending to this log, and you can have any number of readers for this log, right? So maybe one of these readers is my Hadoop cluster I'm trying to feed records into, and maybe another one is some application that's reacting in some way, and another one could be, you know, I don't know, our search index or something like that. And each of these readers is actually pretty cheap. Like, we don't make a second copy of the log for each reader. We just have one, and, you know, they represent their position by, you know, this offset number, which says, hey, If you're at offset 7, that means you've read and processed everything prior to 7, and you've read and processed nothing after that. So this central Kafka cluster really just needs to maintain these logs, and systems will kind of, will track each system just with this one number. And this is almost everything you need to know to understand what Kafka does, except that in Kafka we actually partition this up. A Kafka topic is basically this type of partitioned log. Writes are partitioned up and spread over this. And the reason we do this is to spread these over a cluster of machines. So in Kafka, each of these logs is kind of represented on disk. It's persistent, and it's replicated across multiple machines. So if one of those machines dies, you don't lose your log. You can kind of move these around inside the cluster, so you can start with one machine that has all the logs, you could add machines and kind of spread the load onto these new machines, all without interrupting the kind of consumers or producers of data. And so this is really kind of it, so like if you came to learn what Kafka was, you've kind of like now learned, and that's all there is. It's really very simple. But it turns out this is actually a pretty important data structure, so it kind of unifies real-time subscribers, which will be kind of right here at the end, like listening to every change and reacting right away, with kind of batch or slower subscribers, which might be, you know, maybe I have somebody who subscribes and they're at 5, and then they wake up an hour later and they read up to the end, and then more records get added, and they wake up an hour later and read up to the end, so that kind of like batch loading or processing of data can be done off the same data structure. Just as well. And obviously I've skipped over some details, like we'll have to eventually get rid of or compact or throw away the old data, and Kafka provides different ways of doing that, which are kind of outside the purposes of this talk, but of course you'll have to do that as well. So overall we ended up with something that was really kind of a high-end messaging system that really had the scalability of a file system, so we could, we could dump and write data to it very efficiently. And this turns out to be much more efficient than how messaging systems often internally represent data. So you can usually write to a Kafka machine with hundreds of megabytes per second of I/O per server, and you can add lots of servers. And you can retain lots of data, 'cause they're really just these kind of flat log files with a very simple indexing structure. So you can have many terabytes of data per server, and you can scale that out horizontally. Then the other thing we did that's kind of, you know, part of the hard part is really providing strong guarantees on top of this distributed lock. So when you do a write to the cluster, the cluster replicates it internally across each of these replicas, and it provides back an acknowledgment saying, you know, "Yes, I've got your write, and I won't lose it." And what that means is even if that, you know, server that originally took it dies, any of the new servers that has it is guaranteed to have it. And so this is kind of a typical guarantee in any kind of consistent log algorithm, and there's lots of these out there. But it's relatively detailed to kind of get right. And so we've kind of tried to address that hard problem really so this could be used for important datasets, which is pretty important for the goal we had. So if you had to have your important data going through one thing and then your big data going through another thing, you're kind of losing the purpose of really having the central platform. And then finally, one of the big things we were aiming for is making something that was distributed by default. So something that was really built to run as a cluster. And, you know, a lot of other systems in the space were really built to kind of run a single server, and then maybe there's some story of how you could, like, have multiple single servers. And I think any system, if you've ever used, like, a single server database and a distributed database, It just really is different. You really have to kind of start from the ground up to make something that's distributed. It's very, very hard to tack on the layer, the distributed layer on top, and have it actually work well. And so, you know, what this means is that we kind of thought about replication and we expect that all your data would be replicated because it's not very expensive. So pretty much everybody who uses Kafka uses it with replication. It's not some weird add-on that you kind of configure later with a different system. And it has a built-in model for partitioning and parallelism, which ends up being really important. OK, so this was our kind of solution to this stream data platform. We actually spent significant time productionizing this and getting it to kind of scale up to the type of load we had and rolling it out and getting it to work across data centers and all that kind of stuff. Then we gradually moved each dataset we had onto the system. We started with the kind of user events, which were pretty big and well used. We moved all the operational data next. We worked on database, the set of database changes. This became the basis for a whole really interesting area of real-time analytics tools, so both kind of cool time series charts that would come off of these streams, but also kind of richer domain-specific things like being able to track everything about your kind of complicated service-oriented architecture, which is a problem LinkedIn had. And so we kind of got this whole ecosystem of cool tools, and in particular an area that opened up that we were hoping would open up was the area of stream processing, or applications that kind of directly reacted to events that were happening elsewhere in the business. And this was kind of like a hypothesis we had. You know, the hypothesis was Hey, we don't currently in our company have a lot of real-time stream processing, but we also have no data that's available as a real-time stream. And so you kind of don't know a priori whether the chicken or the egg came first, right? But our belief was, hey, if we had a lot of this data and it was available as a stream, a lot of applications could make use of that. And a lot of the ways we kind of represented and used data could be a lot simpler. So instead of kind of baking everything into the page that served a request, that page would just record, you know, hey, I showed a job impression, or I did this. And anything that needed to respond to that action could just subscribe and do what it needed to do. So if it needed to cache some data or if it needed to update, you know, for job views, maybe you need to update your job recommendations, and maybe that has to go to analytics. A lot of that stuff could kind of come out and happen externally, and these external processes could be added after the fact. So, if the security team decided, oh my God, we have to prevent people from scraping jobs, they could tap into that stream and say, okay, we're enforcing now some limit, and when you've done too many, we're gonna turn you off. And it turned out this actually really paid off, and it led to a lot of, kind of a really enrichment of our understanding of this area of stream processing, which is kind of just emerging now, I would say. You hear a fair amount about it. It's still kind of coming to be in infrastructure. And so I'll talk a little bit about this area and my view on it and what I think is happening, because I think it's really exciting. So the first observation is something I touched on, which is just, If all you really have available is batch data dumps at the end of the day, like if you just have a CSV file that's dumped out at the end of the day, then the only way really to process that is with some kind of big batch analytics process that kicks off and processes at the end of the day. There's no way to process in real time a nightly CSV dump. And this was actually made, we were made aware of this because very early on when I was at LinkedIn, Some company came and talked to us about real-time processing, and we were like, "Well, you know, you have this cool system. We'd love to adopt it, but the only way we have to adopt it is at the end of the day we could take our big files and we could feed it into your real-time system," which is, again, not very appealing. And, okay, so a good example of the difference between kind of batch data collection and batch data processing, real-time data collection and real-time data processing, is the U.S. Census. So this is my— screenshot of the US Census something or other cover sheet. And you know, the way the US Census works is every 10 years people go around with like paper and pencil and they count all the people. So they kind of knock on every door and they say, okay, you're a person, you know, one. You're a person, two. And then they add up all these counts. And of course by the time they've added it up, it's already wrong. And then they do it again 10 years later. And it's kind of a crazy way of counting people if you're like a software engineer where you would really imagine doing something like journaling each birth and death and keeping like a running count of how many people there were. And one of the interesting things about that type of computation is not only does it tell you how many people you currently have in the US, it actually tells you how many people you had at each previous point in time, right? So you have this kind of log of births and deaths And you can actually compute off that the current number. You could use it in a batch fashion where you're like, oh, OK, I want to query it over a period of time. And you could actually run it up to a point in time. You would know, OK, at this particular instant, we had this many people. So it's actually kind of a powerful model for processing or reacting to data. And the second insight we had was that in a lot of ways, stream processing is kind of a generalization of batch processing. It is a generalization of request request-response processing. So by request-response processing, I mean, you know, a web service or a web server or most kind of OLTP databases or key-value stores. Like, you know, I give that system a request and it gives me back a response. So it's kind of one input, one output. And maybe it does lots of these, but it's fundamentally, you know, one goes in, one comes out. And, you know, batch processing, this is kind of your Hadoop jobs or your big data warehouse things cron jobs, you know, probably the first program anybody wrote was like a batch process that, you know, maybe read in a file and printed out something. Maybe that was the second program. I don't know. And, you know, batch processing is basically, you know, take all the inputs and produce all the outputs. And stream processing is actually just a generalization of this. It's, you know, you kind of take some inputs and you produce some outputs. And who gets to decide what "some" means? Well, the program does. And that's actually kind of the power of it. And this is actually not necessarily people's understanding of this area, so when you talk to people about this area, they, you know, some people have an idea that stream processing means, you know, you're like processing really fast and data's just coming, and if it comes too fast, you'll just like throw it on the ground and who knows, like, and then later, you know, in Hadoop, you'll like fix it or something. That is definitely one thing you can do in a stream processor, but it's not, that's not like, a good definition for the area, right? And so I think these two observations, that A, kind of like the computational power of a theoretically good stream processing system could be equal to these two other ways that we have of doing it, and also that it kind of covers this large area of problems, it kind of made me really excited about this area. And I hope it does for you as well. You can actually think of a lot of what happens in a business as a type of streams and reacting to those streams. So if you have maybe a retail store, you can imagine the core data for retail is a set of sales, probably a set of orders and shipments that come in, right? And each of those are representable as a stream of things that happened. And then you have core datasets that react to that. You have an inventory which is computed off of probably orders that arrive. So you have some product and then sales, so the product leaves. And a lot of what happens on the backend where you reorder or change prices or detect fraudulent behavior or whatever, all that kind of stuff, can also be computed off that. So it's actually kind of a powerful way of thinking about streams. And to us, we actually felt like, hey, this log abstraction is really good for this. It's actually something that shows up in all the early academic systems. And so the view of stream processing I have is you basically have a Kafka topic, and then you have some code you write, and it produces data out to some other Kafka topic. And if you've ever, this is actually not that hard to understand. You can think of it just kind of like Unix pipes. Like you have some program or dataset and it kind of streams through this pipe. Not everybody knows that it streams, but it actually, it does. And it streams to some program that processes it and that can kind of pipe out to other things. And the only real differences between that and Kafka is, well, first of all, we don't run all our computer programs on a central mainframe. So it's distributed, right? It works over a network and it's distributed over machines. And then also these kind of topics that Kafka has are multi-subscriber. So if you publish something to a particular pipe, other people can tap into that as well. But the fundamental idea is pretty similar. And how you do this processing is really kind of up to you. Like, you can just write a program that subscribes to the topic and computes something or changes something. but you can also use, you know, a set of kind of emerging stream processing systems that are out there. So the common ones are Storm, Spark Streaming, and Samson. They all work with Kafka, and we're actually adding to this, so if you're interested, I'm gonna be talking tomorrow at a Kafka meetup, if you haven't had, like, enough Kafka, and we'll talk about this stream processing layer that we're gonna be shipping with Kafka in the next release, which is, I think, super cool. But, so all of these things work with Kafka. You can kind of think of the relationship as Kafka has a stream, these frameworks, what they do is really run your program and they do the processing. So kind of like the stream and the processing equal stream processing, and that's as much math as this talk will have. So, like I said, we spent a lot of time rolling this out. I showed the kind of horrific, ugly picture earlier on. Over time, we actually got to something that looks a little bit more like this. So, we were able to capture kind of core data changes coming in from databases, coming in from applications. That became kind of the feed for a lot of the derived data systems. So that would be caches, it would be search, for LinkedIn, it would be newsfeed, that's like all the thingies that show up on your homepage that tell you like, oh, somebody changed jobs. We had internally this OLAP system that did like reporting and stuff and the kind of social graph that says, oh, you know, you're this many hops from that person. So these kind of fed off of these feeds of data that were kind of coming out of the primary stores and they kind of all flowed through this kind of central Kafka hub. And that was kind of like the basic thing was, well, OK, we could connect these kind of real-time systems together. But we could also kind of connect the batch or offline world to the same data streams. And that was kind of the plug everything together story. But the second part that really emerged was this middle thing. So in between come the batch world, which maybe happens on a 24-hour schedule or maybe an hourly schedule or something like that, and the real-time world, which is really kind of you know, as you wait for it. This whole kind of real-time, you know, near real-time asynchronous layer emerged. And so this was kind of anything from, you know, a few hours to a few milliseconds, depending kind of on the application. And this layer, you know, really had two things. It had a set of analytical tools that kind of monitored these flows, and I talked a little bit about some of those, you know, Security was a particular area where people want to be able to prevent bad things usually quickly rather than the next day. And then also a set of applications and stream processors that subscribe to these feeds, transform them into some output feed which went back. And this actually added a lot of power to it because you could take this primary data, you can kind of munge it and then turn it into a search index. Or you could take it and you could munge it and you could turn it into an OLAP store for analysis. and so it kind of made it possible to connect all these different things in different ways. And that usage actually scaled significantly. So the number of messages that kind of flow through Kafka at LinkedIn, I think, is well over a trillion now, which is, you know, very large. And actually, the number of messages read out of the system is much higher because these messages are like multi-subscribers, so most things have at least couple subscribers. And overall, there's about a petabyte of data in the clusters there. This was connected to by virtually every process that would run in any of the data centers. So at least for metrics about the system. And it became the backbone for a lot of the core data stores in different ways. They would basically rely on this commit log to take changes and to fall back on when they failed. And, you know, it's caught on elsewhere as well. So we open-sourced the system pretty early on in the life, and we've been working on it in open source, and so we've worked with a lot of the different companies that were adopting this for interesting use cases. And so you kind of see this across— you know, I don't know how I picked this. There's, like, several thousand companies to pick from, but I kind of chose a random assortment that seemed to come from different industries. And it was really exciting to see this, and so— At a certain point, about a year ago, we actually left our jobs and we started a company around this. We thought, hey, this area of real-time data is probably one of the most important emerging things. A lot of the modern uses of data are inherently very quick. And it's also the area that's the least addressed by infrastructure. The set of infrastructure that exists there is is very weak. And so what we're doing is we're basically packaging together Kafka, a lot of the stuff you have to, you know, have to get running with it, and that's Confluent. So that's what I'm doing now. And if you're interested in this area, there's a bunch of stuff you can read. There's an extremely short book or a very long blog post. The blog post and the book have mostly the same thing, but the blog post is cheaper, as they tend to be. We also have a pretty active blog that goes into a lot of details in, like, areas of stream processing and Kafka stuff, so if you're interested in this area, you should definitely check that out. There will also be a meetup tomorrow, so, like, if you just had, like, a lot of real-time data stuff, but you're like, oh my god, I want more, there is a Kafka meetup, and we'll be talking about stream processing and a bunch of other cool stuff. Unrelated to this talk. And there's Twitter and other things there. And then before I forget, most importantly, there are Kafka stickers. And they're back there and nobody took one. And I was so sad. I think because it wasn't really clear that we were giving them away, but it's by the booth. So if you want a Kafka sticker, you know, they're free. And that's all I got. So I'm totally happy to answer any questions people have.

We have 2 microphones here, so if you do have a question, let the mic come to you.

I can switch off this microphone if you want another one. All right.

In the case of compaction, what technology is behind compaction?

Is it SS table? Yeah, OK. So I kind of like— That's not clear in the documentation. Yeah, OK. So at a high level, I kind of said, oh, you'll have this log, and everybody will append to it. I said, well, at some point you'll probably have to clean something up, and they're like, we won't talk about that. But now you've asked a question in that area, so I should talk about it. So, yeah, so in Kafka there's two ways to kind of clean up the old log. One is to throw away old stuff, and that makes sense for kind of event streams. So if you have a stream of, like, page views, you'll keep that for some period of time. This period could be infinity because you want all the page views for all time, in which case you'll have to keep buying machines. Or it could be like a week or 2 weeks or whatever it is. The other way we have of cleaning up data is what we call compaction, and that makes sense not for something like page views, which in some sense are a stream of, like, inserts, like new events. Every page view is a new event. It makes sense for streams of data that are mutations. So if you have changes coming out of a database, it'll be like, oh, this user updated their account to this, and they updated it to this, and they updated it to this. If you were gonna throw something away, you wouldn't throw away the old updates, you would throw away the old updates for each user, right, for each key. And so this compaction feature in Kafka is really getting rid of the old updates per key. And your question is like, how does it work? Yeah, that's a good question. The way it works is it actually builds an index off the keys and recopies log segments and checks if they're already there and gets rid of them if they're not. So really the impact of it is every time you do one of these compactions, you'll do effectively a full rewrite of the data in the topic minus the things which are compacted away. And there's no concept of memtables, for example? It's analogous to a memtable, but no. So the difference between a memtable or the compaction in Bigtable The kind of compaction that's in like Bigtable or RocksDB or— yeah, whatever. There's a couple of things. So first of all, we always maintain like strict log order for the data. So we'll never like flop rows around. And then secondly, the— you know, like in RocksDB or something, you're really merging sorted segments. And for us, we actually— have like a hash index of keys and will recopy and get rid of it. So at a high level, they're basically doing the same thing. They're compacting, but the mechanism is a little different. I don't know if that makes sense. I think there's a wiki page that describes it a little more. Cool. You said it was like a Git log, and one of the worst things in my life is a Git log when we have a merge conflict, and sometimes it's easier to go smack my teammate across the face than it is to actually fix the merge conflict. How do you deal with, like, do you have to actually have a person when you have a conflict, or how does it, does it do something clever with that? Yeah, that's a good question. I have a feeling it's this microphone. Let me. No? Okay. So yeah, I would say it's actually not as much like a Git log. Probably the closest similarity is a database commit log, which is, like, very similar. They're all kind of the same thing in that they're keeping a list of changes. The technical difference would be databases typically have a single linear history, and Git, because of course the way it works, it has this, like, graph. Of history, and so the merge conflict comes from when you try and merge parts of the graph back into one. For Kafka, you actually don't have that because it is truly a linear sequence of things, and so there is no kind of merging concept really. It would be more like if, I'm trying to think of what the version control analogy would be, but I think the difficulty in version control comes because you kind of check out the code and then you would have to either lock the files while you change them and then unlock them when you check in, which is kind of old school, like RCS, right? Or you have to kind of hope for the best, right? And then fix it when you write. For Kafka, because you don't have that concept of checking out, it doesn't do that part of lock-in. If you think about both these things like a database, Kafka is truly just the log. It doesn't have the kind of lock manager part of it because you don't have that same kind of like read-modify-write cycle. I don't know, hopefully that makes sense. Cool. Other questions? So I was curious how you manage synchronization. How do you maintain a single linear view of all of your data across a cluster? Yeah, yeah, yeah. So just on a per-partition basis, so a topic would be something like all page views at the company. And that topic is actually made up of a bunch of logs, each of which we could call like a partition. And each of these logs is totally ordered, but we don't try and order all page views across the company. In fact, that might not make sense even because they're occurring geographically in different places and there's not really even a total order over them. It turns out that that kind of partition log maps pretty well to, like, the world of software engineering where you have processes which have, you know, or threads which have some kind of, you know, linear notion of time and there's many of those. But then between processes, you actually don't have ordering. So the way you would use it is, you know, you can partition by some key, maybe by user, and then you would have an ordering over, you know, within individual users but you wouldn't have a total ordering over everything. So then your next question might be how do you maintain that ordering within a partition? And, you know, this is kind of like maybe the most studied problem in distributed computers. This is kind of the topic for, you know, like algorithms like Paxos or Raft. If you kind of like vaguely follow this area, you can nerd out on all these algorithms. Kafka is basically doing a similar thing. We wrote a blog post which has some of the comparisons for people who are interested. The big difference between those algorithms and what we're doing is we maintain lots of these. And so a lot of the things like electing leaders and so on, we do in kind of a group way, really for efficiency. The other big difference is we don't require storing as many copies of the data. So everybody has to store like 5 copies of the metadata if you want to tolerate 2 failures. But for your data, you can kind of get away with a cheaper thing where you store 3 copies if you want to get away with 2 failures. Those are kind of like the high-level differences. But, yeah, that's kind of the of the core algorithmic difficulty or whatever if you want to, you know, have many processes which all communicate and data that's replicated over multiple machines and you want to make sure that they all have exactly the same order even if they kind of, you know, fail and come back and fail and come back somewhat rapidly. And you could imagine how you could get like weird stale writes showing up and all kinds of bad things and you have to like protect against all those cases. So, yeah, that's the short answer. The longer answer is you— we wrote a couple of reasonable blog posts on this, so you can kind of go through if you want to nerd out on log replication algorithms, and you can get some idea of the differences and a little bit of what Kafka does. Other questions? Cool. I guess— oh. So from the consumer perspective, right now there seems to be kind of two options available, which is there's the simple consumer where it's like, I want this partition, and I'm going to do all the management myself. And then there's the high-level consumer, which does some magic with ZooKeeper and times out and gets upset and things shuffle around. And everyone tells you don't use it. It's a terrible idea. And so, but I know that there's work on this concept of a new consumer. There's gonna be something that's gonna replace both these. Can you talk about what the direction of that philosophy is going? Sure. Yeah, sure. Yeah, I mean, a lot of people with the high-level consumer, they basically don't understand ZooKeeper and session timeouts, and so then They, like, get frustrated by that. We are doing work that kind of improves the— this is all in the JVM, so if you're in a different language, this is, like, totally inapplicable to you. Kafka supports clients in a bunch of different languages, but this is really about the JVM. So we are doing work to basically come up with a new consumer. I don't know if you've seen the XKCD cartoon where there's, like, oh, there's 15 competing standards, so we need to standardize. Now there's 16. No, so there will be a third consumer. The— it does a couple of things that are new and important. So the first thing it does is it moves all the coordination around who gets to consume what, which is kind of the big thing that the consumer does. It divides up all this work amongst all your consumer processes, so you can do distributed partitioned processing of messages. So it moves all that work onto the server side. That ends up being important for a number of reasons. A, A, you know, the client side is just more finicky. Like, we found people who do this type of messaging, there's usually message processing. There's usually, like, horrific GC pauses, and every, like, bad thing that can happen happens there. B, it makes it much easier to write consumers in other languages, because all the hard stuff is on the server. And C, it allows us to, like, scale up that coordination, you know, very significantly. So those are the kind of the motivations. In addition, we kind of made the API a little bit more powerful, so it kind of does everything that a simple consumer will allow, like go back in time and reread stuff or control what partitions you're assigned or whatever. The— those are the advantages of it. So our, you know, our aim is actually to replace both those existing Java or really Scala clients and— but we'll keep the old clients around really for compatibility, as people are using them for, you know, at least a while. We probably won't add any new features to them. So a lot of the security work that's being done is really just getting done on the new clients. So I don't know if that addresses your question. In terms of, like, the problems with the higher-level consumer, there's a couple. Like, one was, as you added lots of consumers, the time— the time it takes for everybody to agree on what they're consuming got longer and longer. I think that largely is fixed in this rewrite because that whole process of agreement is now just done by whatever Kafka broker is coordinating that group. And so it's kind of like an in-memory computation versus a distributed coordination. So it's fast. And then secondly, you can kind of scale the number of these coordinators with each broker, like each broker has one, so if you have 100 Kafka machines, you get 100 coordinators. So, you know, I think it should address many of the problems, although fundamentally anything which is gonna automatically tolerate failures will have something in it which says like, oh, if you don't do anything for X amount of time, I kick you out, and it turns out there's really no perfect value of X because everybody will kind of pause and not quite be dead and then come back, and so if you get these processes which kind of pause and die and then come back and then pause and die and come back and kind of churn, you'll always have this kind of interruption of that person joining and leaving the group. And so that fundamental problem probably won't be resolved, but I think most of the incidental ones will. Hopefully that answers your question. Cool. Hi. First of all, I want to say Kafka is amazing.

I've been working 3 years with it. It's truly amazing. Tons of questions, but I'm going to ask two about roadmap.

OK.

Independent producer transactions, is it going to happen ever?

Because it's like proposal for a year now or something like that.

And second question, when are we going to be able to delete topics finally?

Oh, OK. That's good. OK. So delete topic should work now as of the latest release. Several people have claimed it didn't. It turns out they just didn't turn it on. So if you believe it does not work, you should definitely report a bug. Now, I realize in the past we said Delete Topic worked and there were corner case issues. And then we said, no, it really works. And there were still corner case issues. But this time we're saying, you know, it works. No, and we've said that for a while. And nobody has come up with a corner case issue that's reproducible, that, you know, blah, blah, blah. And so there are many people using it in production. So we think, we think now it works. So if you find a corner case issue, definitely tell us. You know, we will sob quietly to ourselves. But, but it's good to know about. It turns out that deleting data in a distributed system is much harder than writing data. And yet usually, like, losing data is so easy. I don't know why. You know, it kind of makes sense, because, like, if you delete something and the server that has the data is down, and then it comes back up later, and then somebody could be recreating, you know, the topic with the same name. Like, there's a lot of corner cases. Didn't think about it at first. Well, anyhow, now we thought about it. Okay, so, all right, that was your second question. The next was the transactional producer. So the— what's a transactional producer? The idea is, you know, currently when you write to one partition, if you get an acknowledgement back, you know your messages were written. If you don't, you don't know. And that is really something which happens, you know, per partition or per server. And so there's no guarantee if you do something like, you know, write to this topic and write to that topic and write to this other topic, you don't really know that they will all happen or not happen together, right? So it could be that some of these happen and some of them you have to, like, keep retrying until that server is able to receive it. So the concept of a transaction was to provide some kind of guarantees which are analogous to a database. The analogies to a database are kind of close, 'cause a database also has a log and similar things. And so in particular, we want the writes to all be atomic, like happen together or not, and maybe have some kind of ordering between transactions that makes sense. So we did a relatively complete prototype of this about a year ago. For a long time, we were a little bit throttled by the fact that the Kafka team was, like, 4 people, and so there was, like, only so much you can kind of productionize at a certain rate. Now we have a lot more people, so it's kind of slowly getting faster, but of course it takes time for people to come up to speed. So we are going to seriously address this. We'll probably start work on that probably the beginning of next year where we actually have people working on it. Assuming we go the direction we kind of already explored reasonably well, you know, I don't know, it may take us a quarter or so to get that done. If we kind of redesign it as part of that, you know, it may take longer, but we're definitely doing it. The primary goal of that is really to support stream processing and making it really easy to do these kind of processes that read and consume messages and produce output and make sure that they get, you know, kind of exactly-once guarantees, people call it, but it's not really a very good description, but for that type of processing. So the answer is, yeah, it will happen. It's not in this next release. It's definitely feasible. Like, we did a relatively complete prototype of it. So it's doable. The performance impacts aren't too high, blah blah blah. But just nobody has been able to work on it. It was one of these intern projects, if people have ever done intern projects. So we got these really smart, you know, PhD students to come in and do it. But it's very hard to take that and kind of productionize it. I don't think we ever solved the, like, productionizing PhD student work except by hiring the PhD students and having them do it, which works actually quite well. One of the people at Confluent wrote the first version of the consumer that we're just doing while an intern, and then we made him come finish it. All right, one last question? Yeah. Cool. Yeah, I had one. It seems like as you guys come out with more and more releases, you seem to be relying less and less on ZooKeeper for a lot of the functionality. I was wondering what kind of the grand vision is for that and if there might be a future of Kafka without ZooKeeper or kind of where that might be going. Yeah, that's a great question. So there's kind of like different camps, I would say. So like some people hate ZooKeeper but they like something which is almost exactly like ZooKeeper. But less mature. Those people I actually don't totally understand. I think they just decided on that thing, and so you kind of want it to use whatever thing you have. My take is replacing Zookeeper with something kind of almost like it but different probably isn't a huge value-add. However, there's two things we'd like to do in kind of a short-term, long-term. So as of this upcoming release with the new clients, there's no connection between the client and Zookeeper. So the clients never interact or connect with Zookeeper in any way. And so we've actually been working on this for well over a year to basically move every usage of Zookeeper in the clients off. So we did, you know, we redid how your offset or position in the log was stored to kind of help scale that, and that moves it off Zookeeper. We redid kind of how you discover metadata about the cluster, and that moved it off Zookeeper. And this last thing we're doing that I just talked about for the new consumer. That's about how work is divided up amongst the pool of consumers. So that's kind of like the final piece of that puzzle. The reason for that is it turns out that having lots and lots of processes that all connect to ZooKeeper is not a good thing to do. I mean, in my, like, humble opinion, like, a lot of people kind of talked about ZooKeeper as something you would have everything in your data center connect to and you would coordinate it all. That just It turns out not to be good. For a bunch of reasons, it's just hard to manage if you have a company with more than 5 people. You don't know what they're doing and they'll do something abusive that hurts ZooKeeper and blah, blah, blah. So we're basically making ZooKeeper, as of this next release, just an implementation detail. So next release should be, you know, in mid-November. So as of that point, you know, Kafka relies on ZooKeeper, but your clients don't connect to it. So you don't have to secure the connection to it if you're trying to do security. You don't have to think about, like, hardening it. Like, Kafka's adding quotas and stuff to kind of protect itself against abusive clients, and that won't be an issue for ZooKeeper, but it'll still be there. You have to set it up. I actually think there would be a lot of advantages to not having a second operational dependency at all. So whereas, like, replacing ZooKeeper with some other distributed coordination system, I think is not that compelling. I think it would be compelling to have the Kafka cluster just kind of stand alone. Now, you also want it to work really well, and ZooKeeper is like a pretty mature piece of software, so we haven't started that yet. We don't have a design for getting rid of it entirely, but I think it would be compelling if you didn't have this other operational dependency, you didn't have to set up a bunch of ZooKeeper machines just to get started, I think that would be pretty good. It also turns out that what Kafka does in terms of maintaining a kind of consistent log is very similar to what ZooKeeper internally does. And we use ZooKeeper properties to do some of our stuff, but it would be possible to kind of fold that in in some way, but it's kind of undefined and not scoped out. So that's probably the complete answer. In the short term, it's kind of like an implementation detail. Clients won't talk to it, and I think that addresses most of the kind of like operability Zookeeper problems people had. Longer term, yeah, it would be interesting to explore how to just, like, do it natively inside the cluster. If we did that, it would have to be done really, really well, and we'd have to be really sure that it was solid. But the advantage of that would be kind of easier to get started out of the box, which I think is really important. Cool. All right, thanks, you guys. I'm super happy to take any other questions.

Yeah, I saw a few more hands raised, so yeah, feel free to stick around. Jay will be here to answer questions.

Yes?

Yes? There's snacks, there's drinks.

We're not going to kick you out too soon, so there are stickers.

Thank you all for coming, and thank you, Jay Kreps.