SBTB 2023: Tim Berglund, Pinot, Why Are You So Fast?
Recording: SBTB 2023: Tim Berglund, Pinot, Why Are You So Fast?
[Music] hey thanks for coming everybody uh I would like to talk to you about Apache penino Apache Pino is a real-time analytics database it is as the label says fast and I want to answer some questions about why it's like very it's very interesting how you engineer an analytics database to answer in milliseconds rather than seconds in my opinion even more interesting to ask why would you want a new database and I think there's a bit of story to tell there uh so you know who's telling it my name is Tim I live down in Mountain View and I work for a company called star tree we make uh cloud service fully managed cloud service based on Apache Pino uh I run the developer relations team there it's worthwhile to just I think have a brief history of I don't know Computing we only have 30 minutes here so um I'll make it brief this is an IBM 1401 uh about 5 minutes from where I live in Mountain View at the Computer History Museum they have two working 1401 machines I saw one of them last Saturday I won't say that I didn't weep a little I won't say that I did but I won't say that I didn't a little bit it's really an amazing experience uh to see one of these things running um and it's it's was kind of funny to note you know that the programs are stored on Punch Cards input data is stored on Punch Cards and the pain in the butt of getting Punch Cards read into the machine I was thinking yep inest is still actually that painful nothing has changed in 70 years these were batch analytics machines okay this is the the dawn of commercial Computing 1959 uh very few companies governments were were were Rich enough to Warrant buying a computer but those who did did analytics on large batches of data it wasn't until uh this product pictured here you might guess is the mid-60s 1964 the IBM 2260 was their first standardized terminal product and so now uh you by this time computers are fast enough there are interactive terminals you type you see characters you provide input you see output and the idea of recording a transaction in a computer becomes a thing so it takes a few years for the idea of a transactional database even to emerge we start out in this batch analytics mode and now all of a sudden we are uh recording changes that happen in the world live in a transactional database well and if you wanted to know what was true if you kind of wanted to do analytics on that transactional data well it was all in one place reporting was just a feature of the software there was no analytics as a discipline there was a report that you wrote somehow um another funny detail I I was looking for a stock image of green bar that's harder than you might think they actually have boxes of it at the Computer History Museum so if you're from out of town Mountain View is about 45 60 minutes away it's a bit of a hall strongly recommend you go check this out anyway a few years later mainframes take over you know Computing uh mini computers emerge which are kind of architecturally the same they're just cheaper but then in the early 80s we have this we have the PC so now Computing is cheap enough where you have storage you have IO uh and you have compute all in a cheap enough package that you can put it on the desktop of a manager executive knowledge worker uh and that is a GameChanger so you have from the 50s until the 80s essentially I'm going to argue an unbroken Paradigm of you think you've written a monolith you have no idea the the monolithic Mainframe software now all of a sudden you've got programs running on PCS and where does the data live in some closet somewhere on a departmental database server now all of a sudden this is a thing this caused a crisis in analytics by the way I was a nerdy teenager reading bite magazine when this Revolution was happening in the 80s the client server Revolution okay uh and this was a revolution which was to have lasted for a generation we are only just now seeing it it come apart and uh now that all of the data in the Enterprise is squirreled away in all these these these databases everywhere all of a sudden their decision makers can't get one view of the business anymore as what do we do the crisis or the the the Revolution and application architecture from Mainframe to um uh to client server creates a crisis in analytics or reporting as it was called then which gave rise to the data warehouse late 80s early 90s this is emerging as a discipline books are written Wars are fought uh consultancies created and uh this whole discipline arises now the technologies that we were building data warehouses out of in the '90s and early 2000s are definitely considered Legacy now but uh that parad just like the client server Paradigm was pretty stable because what happened a few years after this some of you are old enough to remember well the web happened all right seems like a giant revolution in the way we access computers and the way Society works I think that's fair but honestly the people who were building the early web were people who maybe 10 years ago were helping lead the client server Revolution so how did you build a website well you picked up your tools the way you built software which was client server and you made websites out of that and so that same Paradigm incredibly survived the the onset of the internet the internet was built with with client server applications and data warehouses kept ticking along after a little while say 2010 we started to realize okay we've got a lot more data from these websites just a very different scale that we we you know recording user populations and building out application uh databases that that the old style data warehouse stuff doesn't work so big data happened um and this by the way I think I asked Adobe Firefly to show me a picture of big data and I just I like it so much I never want to stop using it it's like an electronic elephant I think that really works so this was definitely a new set of tools to understand to to do analytics and understand what analytics was all about there were real changes the that that that were ushered in uh all new tool sets but I'm going to suggest maybe not everything really changed as much as it might have because you still had this idea that there's that application data over there we're going to get it at some point and we're going to put it here and then we're going to do analytics on it over here so uh it's a lot like a data warehouse which is why we just changed the word from Warehouse to Lake and that reflected some interesting changes in approaches to analytics but it was still the basic idea that we're going to get all our data we're going to put it there we're going to do stuff with it okay so again massive changes happening and we're still kind of carrying this artifact this Paradigm of the way we used to do things with us around that same time we started to use the word monolith as a bad word this single application centered on a d a single database that starting to be hard to test hard to deploy hard to scale and uh crazy people started telling us no you should break that program up into 100 or 50 or 25 separate programs and have them all talk to each other around 2010 2012 that was miserable didn't work very well at all until what happened a few years ago it seems like we figured out a good way to do microservices Kafka entered the scene and this notion of the event-driven application a set of services talking to each other over Kafka topics seemed to work pretty well what happened here was the the database at the center of our world kind of stopped being you know the primary thing and the the primary place where data live became a Kafka topic this is happening right now it is I'm going to argue the same kind of thing that happened when the PC became a thing in 1982 and that client server application Revolution the way business software was written utterly changed and a whole new tool set new stack new set of programming capabilities everything emerged this is happening to us right now uh as as we go from monoliths or the client server Paradigm that we all grew up with into event- driven architecture just like before when that application architecture Revolution ushered in a new way of doing analytics that's kind of what's setting the stage for things like Pino now I want to tell you just a little bit about how Pino Works talk is called why penino why are you so fast so we'll talk about why it's fast how it's built uh and I want at the end to just fill in some some holes in this story and let you know where I think we're going and why penino matters now all right well Pino's a database it uses tables it's really not trying to impress you with any new and unusual data model you know they're tables it's a it's a it's a big box you put data into but because it's a distributed database it breaks data up into segments ments or how data is stored on disk and even if you're not you know operating your own Pino cluster um you kind of will have to think about segments and care about them a little bit so it's worth diving into that a little bit a segment you can think of as maybe like a 200 300 400 megabyte chunk of of data there a bunch of rows in uh the table however many rows there are so you can have as many segments as you need and we'll we'll see how they get distributed later let's look at how they're built though because if you're going to be an analytics database and not a transactional database you're going to organize your data on disk differently if you're a transactional database you're primarily interested in things what's a whole row right I want to create a row or read a row and you usually want a substantial portion of that row all at once and so you serialize all the bites of that Row in one place in storage when you're doing analytics you are much more likely to want to scan a number of measurements a single column you want to get a bunch of those off disk in an efficient way because probably you're scanning and aggregating a bunch of values so imagine this data I've got some little chunk of Json looks like clickstream data or something when we create segments we're going to take maybe you know 100 thousand few hundred thousand of those rows and all of the column values we batch up together so all of the values of the IP column the user ID column remote user time Etc all those get stored together so columns are stored together on disk uh so this this is actually what it means to be a column or database so you're to zoom in on one of those like bytes bytes might be a good value that we might want to compute an aggregation over if you look at that chunk of the segment you would see a bunch of individual bite values lined up together so at no point in a segment is the whole is a whole row stored all at once we've grouped up batched up columns together that's an optimization for being an old app database Pino is in complete control of its own file system by the way sometimes people have asked me what's it used for storage is it is it paret or something and no I mean there are systems that you can use to query sort of arbitrary chunks and and lakes if you will of data on uh in in cloud storage but that's not Pino Pino has Direct Control of the file format tightly storage that's tightly coupled to compute all that's in one place and that's key to it being as fast as it is so we got all these segments lots and lots of segments just give them numbers they go on a machine called a server as you know naming things is hard this is not a good name it's not the only component in Pino's physical architecture but we call it server uh so you have multiple servers this is a component that you can scale arbitrarily uh and servers are the place where the the real work gets done uh the oops the actual um storage of data and compute involved query processing is almost entirely done on the servers clients don't talk directly to servers clients talk to Brokers what Brokers do are uh what Brokers do is they take a query in uh they'll parse it and they'll try to figure out which servers have the segments implicated in that query so there are going to be predicates that you're going to have and those predicates might provide hints about what segments are involved and that might let you um prove the number of servers that you have to scatter the query to hopefully it will if it doesn't then you know okay you can just scatter it to everybody but you'd like there's this interesting trade-off that it has to make between um as want wanting to involve as few servers as possible but also wanting to parallelize the work so query routing is actually a fairly sophisticated and performance sensitive thing and it's another one of those things that that takes place in the broker that's responsible for Pino being as fast as it is so the broker scatters the query it says hey I need segments two and8 and so I'll just find those servers that have those segments send the query to those servers the servers will actually perform the filtering and scanning and aggregation and send intermediate results back up to the broker in this single stage query example which is sort of the default for how queries work in Pino um what's probably coming back out of the server is either a scaler right if you're Computing say an average of something well that's a that's a number that comes back or if you're doing a group by it's a set of key value pairs of relatively low cardinality so you don't have massive amounts of data coming from the server back to the broker the broker is then going to effectively reduce those into a single result result set send it back to the client uh there is an option videos available online I won't give you animations today but uh there is a way in contemporary Pino to the broker's compute a multi-stage query involving various stages of servers for more complex things like joins and and SQL window functions and things like that those are Now supported in Pino uh and that plan looks a little more complex than what you're what you're seeing here but this is kind of the basics of a single stage query and for the basic filter and aggregate bread and butter analytics query this works great um overseeing all this is a machine called the controller the controller's job is to orchestrate certain operations like loading new data connecting to Kafka to inest data and making sure that metadata is distributed so the Brokers uh their primary responsibility is to keep track of cluster metadata keep that picture in mind so they can do good query routing so that Pino can remain fast the controller is the one that oversees that metadata storing it all persistently in Zookeeper as we do there's this other little component off to the side called minion um and minions perform outof band maintenance tasks that that aren't queries you don't want them running on servers but uh very common and and almost necessary thing that minions do is loading batch data if you've got a giant collection of say CSV files in an S3 bucket uh there's going to be a job that runs on a minion creates those segments and loads them into the system and that really is the physical architecture of Pino right there the basics of the components and how they work together to make a database that runs fast and everything about this system is engineered to provide results in milliseconds not seconds and there are real life use cases where you have actual queries of large sets of of user data returning results in under 10 milliseconds more commonly small numbers of tens of milliseconds and to do that you have to have control over your storage you have to have compute and storage tightly coupled you have to have compute and storage pre-allocated you're not going to go uh find a place for a job to run when you've got query compute to do you're going to that's dramatic um we have we have an AV AV issue you're going to send the query to exactly where the data is and that compute is on standby ready to go another way youo as fast is indexes I'm not going to tell you about all of these but these are the main ones um some common things like uh there a couple different kinds of text indexes uh range in index for range queries A specialized kind of range index for uh queries involving timestamps all things that help you filter in in a very a very efficient way when a query is being executed on a server so indexes exist at the segment level and so you have a query down on the server that compute happening indexes help you read less off of dis that's the idea uh since really when you're when you're scanning storage you can only make that go so fast your next best option is to do less of it and indexes help you do less of that I do want to show you how one index works because it's kind of cool it has a cool name and it it it gives performance Benchmark numbers when when people do like bake offs with it their first response is usually they think something is broken because it can't possibly be that fast right it has to be some sort of mistake result must be cached or something like that that is the star tree index and it's hyphenated and I have to say it differently so you know it's not star tree the company that I work for we just took our name is an implementation detail that leaked into the interface because we thought it sounded cool um but the star tree index is is not to be confused with the name of the company now imagine you have uh we'll say some ad data there's not a great deal of adtech stuff that I see being done with Pino but there are a few kind of interesting and and people usually think of adtech as as somehow like a dirty business um we did a about a year ago a nice video with a guy who ran engineering for a a company that used Pino to do adtech and it was like really heartwarming adtec you know like so you got your recipe blog and you want to be able to sell cooking gear on your recipe blogs it's kind of mom and pop adte you know very heartwarming let's just pretend this is that and the query that we want to run is again the bread and butter filter and aggregate I've got three predicates those are country browser and local I don't know which one of those three predicat is going to throw show up in the query some are all of them may and I've got one measurement and that's add Impressions and I want to aggregate add Impressions we'll say sum I want the total of add Impressions based on those predicates what the star tree index does and this is the only time pre-computation happens in Pino Pino does not ask you to pre-compute anything it doesn't do any pre-computation except when it builds this index it builds this tree and let's say you want um I don't know all of the uh Spanish language Impressions right that's it's that's just uh local equals es is the only predicate that shows up in the query what are you going to do well you we'll Traverse this we'll go from the root node and we'll pick the country country is the next level down well wild card for Country we don't have country and then browser is the next level down we'll pick wild card for browser and then under there we see oh well 500 is the pre-computed aggregate for the Spanish local it's right there in the table the index can contain I can Define other Aggregates on other measurements as long as this is my set of of um uh of predicates I can have as many Aggregates stored in that note as I want and you get essentially login access to that thing so it looks like a hack this is where you can get on petabyte data you can get tens of milliseconds on real live aggregations and you don't have to do any that pre-computing um let's skip a little bit to ingestion I'd like to tell you how ingestion works because that's part of the Kafka story and then we'll finish up all right um this comes from an example on dev. Star tree. um our developer site this is a a recipe that's in the documentation imagine you have this gigantic collection of CSV data you'd like to import there's a procedure for how to do it and if you wanted it to be more CSV data than that that's okay with us um the basic idea is you're going to read this into a table right and like I said before as you're reading batch data into a table you're creating segments and loading them into that table well what does that actually look like um well this data you create two files and they're both in Json you don't ever type create table or anything like that you create a schema file in Json you can reuse that between tables that's one schema definition uh that that has your Dimensions your metrics and your date time Fields those are the three kinds of columns in there and then and you don't need to look at the details of this obviously you can find these online uh you create what's called a table config file that table config Pig file gives the table a name points to the schema and creates whatever other configuration you want If This Were A create table uh statement you'd have you know would say options and there'd be that big bulky set of key value pairs it would be sort of ugly so here we are and then you create the table the command line command and if you want to ingest the data the only place you're going to see Amel in Pino you're going to create an ingestion job spec that looks like that run another command all of that to say that's a minion job remember I I mentioned those little minions down there well that kicks off a job that gets distributed to one of the minion workers it has access to the input file and uh it says okay we'll select you for that work it reads in the CSV and actually locally creates that segment remember that segment is like a 500 megabyte is file of of input data obviously you're not going to get 500 megabytes from that CSV file but your data is larger and the minion sends it to the controller and the controller distributes it to a server and then it's loaded and available for querying so that's kind of how batch data works now streaming data is really what this is all about because Pino came to life was actually created at LinkedIn three years after Kafka was created so all of a sudden they had all these streaming data pipelines they had to come up with ways of doing analytics on those for users so they wanted live olap queries against live data just ingested from Kafka that could be in response to a page load or a tap on a mobile app so something that's fast enough that I don't feel like the app is broken or the site is broken it's you don't have 15 seconds for a spark job a fast spark job to run it has to be right now uh and so Pino was created for that now what happens here is uh something very similar except as I'm consuming from a Kafka topic uh there's an inmemory data structure called the consuming segment and that is that is queriable by the way so as soon as we ingest a message from Kafka the query routing broker server configuration all that we'll be able to query that that consuming segment so records show up in queries immediately as soon as they're consumed but of course it's in memory so once it fills up we flush it to a real segment on disk through that same segment distribution process and then that on disk segment data is available for query just like any other data so that's how Kafka ingest works as soon as a message is ingested it's available for query and there are some cool uh there's some more details to this um number one this is kind of interesting in the table config file you actually specify bootstrap servers topic name credentials any anything you need to connect to Kafka the table is configured to connect directly to Kafka once you create that table it begins consuming from that topic from the start from the end however you set it up so uh a so-called real time table in Pino is a thing that is connected to a Kafka topic and and ingests immediately and what actually happens here is the controller will say okay here's this new table I see in the metadata uh you know you Mr server one you're going to connect to uh this Kafka topic and you're going to consume directly from it so if you know Kafka there's no Kafka connect here there's no uh middle stage of any kind uh the servers themselves act as Kafka consumers and so it'll consume fill up the consuming segment flush segments and so on in fact it even gets a little better than this to where servers can do this in a partition aware way so if if you've got a query that uh uses as a predicate your partition key sometimes this happens right however those cka topics are partitioned uh if if that shows up in a predicate well I can make sure that every message with that key that is out there in that partition is always going to be on the same server asterisk almost uh you know we're going to do our best to make sure those segments land on that server and the Brokers are going to be aware of this so we get to do pruning and we get to make penino very fast we get to use that as a little optimization when it's available because again the the servers themselves are are Kafka consumers directly so there's all kinds of things built in here uh to make this fast to make it uh I I've always described the the kfka integration as a very intimate one it is Pino grew up alongside Kafka when Kafka was you know barely learning how to replicate data Pino was is being created uh and has has very much grown up with it and it's it's uh a smart match I don't often describe penino this way because if if you don't know what penino is if you haven't kind of been through this little half hour explanation it's not helpful but one can think of Pino as queriable Kafka that's not a complete account there is more to it than that but remember my my history right I said we started with these batch analytics machines loading data in Punch Cards suddenly the idea of transactions became a thing reporting was still easy until it wasn't because there was an A revolution in application architecture client server began to displace mainframes for most new development and so we had to invent a new way to do analytics that's the data warehouse the web didn't break any of this right away it took 10 or 15 years but then it started to break it and we created Big Data because we wanted even bigger data warehouses and we realized wait the way we've been writing applications has to change and that got us into this event-driven world that we're all living in and before this transition we kind of knew how to write applications right there was a database and you had a language you know you going to use Java or whatever it was maybe some you had some rough circumstances and it was schola you know anything can happen right you pick a language those languages back then and you've got your database you maybe pick a web framework or you know some limited number of choices to build that application everybody knew how to do it that the the choices were put out it seemed like it was hard but it wasn't they were all right in front of you now you've got application code that you're writing sure this is my stuff in microservices and at the bottom down here of the stack there's a bunch of Kafka topics but in between it really feels like wild Uncharted Territory and this this crazy land land grab for what is going to be happening in the stack there it's a hard time to be a developer during this time of transition this would have been like if you had been a Mainframe developer in 1985 and you got a job doing client server things everything is different you don't know how to do anything and this is where we are right now we're trying to figure that space out one of those places one of those new parts of the stack is Pino we have to be able to look back at these streams of events that are now ruling Our Lives lives and ask what's happened and get insight into what has happened that's fundamentally what an analytics database does this is one of the parts of the stack that I think we're going to find is critical if you want to know more join me star tree Community slack you can slack me you can ask other people point you to other resources got lots of videos sample code documentation everything you need thanks for being here question are we doing questions I love questions yeah okay tell me when to stop I'll stay here until the panel uh thanks for the talk um you mentioned at some point that the storage is tightly coupled to the compute is that makes sense that that's how you would start is that what you guys are sticking with and is it by Design is the sort of an assertion that you can't get the kind of latency numbers you want by decoupling storage just curious what good question I like almost planted but not um and if you will pardon me I will Veer off into commercial territory because that's where the answer exists right now in Star tree Cloud um the managed service we have a tiered storage extension and if you go uh YouTube and look for star tree tiered storage you'll see a video where uh the engineer who led that effort her name is nah po she and I appear to be writing in thin air talking about how it works it's very cool obviously it is slower so um parts of segments can now be stored in S3 in Star tree cloud or you know Cloud blob store of your choice of course it's slower it's a lot less slower than you might think it's it's there's some clever things in there so yes of course we're going to negotiate that and that'll that'll you know leak into the open source product at some point you know these things go awesome um this is great especially for uh the notion of Pino for customers that particularly use Pino production what would you say are there statistics for like how big that the cluster scales and you know just the overall magnitude of the the growth of Pino as a whole as a stack so how how big are there's like a particular customer use or scale P to the the furthest today that we know um yeah uh it's funny you say customers I'm I'm not on the customer facing side of things in Deo I don't I don't uh I don't make so sales I try not to no that's a joke I love sales typical production deployments are you might see tens of terabytes hundreds of terabytes you'll see some petabyte deployments um and some folks have presentations that that we've got on the website videos where they they talk about those but yeah many many terabytes small numbers of pedabytes okay uh I think that might be it for questions um table two outside afterwards table two yeah um Q&A table two if you have further questions oh I I go to table two yes I please please T I would love to see you at table two thanks for being [Music] here