CMU Database Group · 1:22:55 · https://www.youtube.com/watch?v=L6E75IyBgpk
0:00 Official members bounce bust off the chrome. Realize it's real hell where we call home. Official members bounce bust off the chrome. Realize it's real hell where we one. I vision fat licks and m clips. The chrome make the average imitator do back flips. You lack a pistol. coming official.
0:29 Today's going be the last real lecture uh that that'll be covered in the the final exam. So, it's sort of like finishing up the thing of the semester. Um the last year that we cover the final exam, next class will be uh prepary session. And what I decided to do this year instead of having you guys vote for something um I can click go. instead instead of having you guys vote for like what systems to cover, since I'm not teaching 721 next semester, I'm going to try to cram 721 into a single lecture
0:57 and and just go through like click house, big query, uh snowflake, data bricks, um you just try to go yellow brick, try to hit as many systems as we can and you'll start to see how we synthesize again all the ideas that we've talked about the entire semester, we now start putting it into real systems. So again, you'll be in a position that that when someone comes along and says, "Hey, I have this new system. It does amazing things." and you go look in their documentation, see what they're actually doing, you realize, oh, it's this, this, this, and this from the things we covered this semester.
1:21 It's not, you know, some it's almost always never going to be uh something like we haven't looked at before, just variations of it. All right, so again, that'll be Wednesday's class. Uh, and we'll also do the final final exam uh review in the beginning of the lecture. Project four again is due on December 8th. Homework six will be due on December 9th. Uh, we'll have the we'll have the I'll announce this next class.
1:44 have the Saturday session uh on the 7th for the TAs uh on the Saturday before the Sunday when project's due we'll announce that but it'll be the same time same location as before final exam will be on December 13th uh and I'll release guy with the practice final exam on uh on on tomorrow post that on PIA any questions about any of these yes is the final exam cumulative or is it like the second midterm question is the final cumulative uh no it'll be it'll cover the material we covered since the midterm But obviously there's things you need to know that we covered before the like if you forget what SQL is, you're going to have problems. So don't don't do that, right?
2:21 You got to know what a buffer pool is because we'll actually see that today a little bit as well, right? But am I going to ask you like what's the complexity of this like from something in the beginning of the semester? No. Other questions?
2:32 And then it's it's a three-hour slot of time, but it'll be the the number of questions will be roughly same as the midterm. If you want to take the whole three hours, go ahead. I don't care. Okay. And then I don't think this is anymore in the class, but if you have uh need accommodations, please contact me as soon as possible. Make sure we go through the office and get all that set up for you ahead of time. Okay.
2:52 All right. So, today uh we have we have another speaker uh and is actually very relevant to what we're talking about today, but object stores. So, there's a system out of China called data bend. And then they forked off this piece of the of the the architecture that that just abstracts away what the storage looks like going talking to object stores which again what we'll see today in in OLAP systems. So he's giving a talk today at 4:30.
3:14 The Grep Time DB guy overslept last week. I felt bad because it's 4 a.m. in China when he was calling in. Uh so he missed his alarm. So we're giving him a second shot next week uh to try to try to come give the talk again. So this is optional but again it's starting to tie in all the things that we've been talking about the entire semester. All right.
3:32 So today we're going to talk about distributed analytical systems, right? Last class was distributed transactional systems. Now we're going to see what we're going to do differently in a database system architecture when we now have to spread across multiple nodes. But before we get to that, we sort of understand what these sort of large scale analytical environments look like. Like where's the data coming from? What does the data actually look like? And so we we sort of showed this diagram early in the semester where we said a typical setup is that you'll have your front-end ultimate databases, right?
3:57 These are the things these are the these the application that's talking to the outside world like whatever your phone app is when it communicates to to the web server right it's going to be usually hitting up one of these uh transactional database systems and then the idea is that you want to get all of your data from these these different data silos the LTP systems into a single data data warehouse or OLAP system so you can do analytics across all your data and not just a subset of it and so we said that there was this this uh intermediate layer here typically They call it extract, transform and load or ETL where the idea is that it's a some kind of middleware framework that knows how to go to these systems get data out of them and shove it to the OLAP system or we saw change data capture where you look at the right head log entries that'll be written to by the database system and you just convert them to data records that you then insert into the data warehouse. Right?
4:46 This ETL piece is keeping trying to keep everything in sync. Right? There's this there's an old topic. There's a bunch of tools in this space. uh Informatica and Five Chan Pi the most most famous ones but Amazon has glue there's there's a bunch of these things but we had actually a speaker early in the semester uh from from dbt and what they're actually doing is slightly different so instead of ETL it's ELT extract extract load transform
5:12 and the idea there is you instead of doing uh some computation to change what the data looks like before you put in the data warehouse you just throw everything in the data warehouse right because it could just be a bunch of files on and park paret files in S3 and then you would use something like dbt to transform them, convert them into sort of curated data sets, right? DBT isn't the only one, but it's it's certainly the most most famous one in this space, right?
5:35 And so what's different about this and what the DBT guy we were sort of joking about like they don't care if your queries you write a bad query because it's going to run on Snowflake, whatever your back end data warehouse is, they're not paying for it. like all the transform pieces happening inside your data warehouse because these systems are scalable and can handle uh processing and crunching uh large data sets. Right?
5:56 So, so again, this is a very very common setup, right? An example I like to maybe use is like Walmart basically has uh bunch of databases. I think uh that may not be public, but there's every every store has has a database, right? Um and and so so you sort of think that it's like a data center like within one store there's there's a database to tell you what what people are buying, but eventually you want to get that back into the data warehouse to a centralized location.
6:25 So you know across all your stores what people are buying and then you can start doing uh extrapolate new knowledge from this data you're collecting to make better decisions how you run your organization. So a a a a well-known uh use case was Walmart wanted to know what people were buying before hurricane hit and after hurricane hit. And so they would look at all the sales records from all the stores, put in a giant data warehouse,
6:51 figure out, okay, within these regions, these time ranges, uh, uh, times of the year at at at, you know, in Florida and wherever else hurricanes hit, what are people buying ahead of time, so you can stock up before the hurricane hits? And then what are people going to buy immediately after, like, I don't know, blankets and things like that. And so that they can stage all their trucks right outside the danger zone of the hurricane, but after the hurricane hit, they can then flood the stores with all the things people need to buy, right?
7:15 That's an example of using you know using data to extrapolate to find new things to do make decisions how you run run your uh your company. So these are typically called decision support systems, right? That example where I said like instead of Walmart guessing what people buy before and after hurricane, you just look at the data and look what people bought and then then make decisions based on that. Right? So this idea is not new. It goes back into uh
7:40 late 90s early 2000s uh and back in the old days again it was only the Walmarts of the war Walmarts of the world like big banks big companies that had large data sets. now it doesn't take very much of you know slap up a web app and get a lot of people using it pretty quickly uh and start collecting a lot of data. So the the proliferation of these sort of large scale data analytical tools has uh has has only increase because there more people actually need be to do these things.
8:06 Now I don't think we talked about star schema, snowflake schema, right? No. Okay. So when we looked at the OTP systems, we sort of talked about how the the schema would look as like a tree hierarchy like you have you have all your orders. You have Andy's orders and Andes has has the order items and the items that they purchased, right? Sort of this clear hierarchy between uh entities within within your data set. But that's actually not going to be great for analytics because now when I want to go start, you know, start processing my data and figure out what people are buying at, you know, at at historical trends, uh, I got to do a bunch of joins across that tree structure, and that's going to be expensive. That's going to be slow.
8:41 So there was this moment in the late 90s 2000s and this is actually pretty common today where you would organize the data in your data warehouse different than how you would organize the data the schema if you will of of how you organize data in your in your transactional systems and the two major approaches are star schemas and snowflake scheas.
9:02 So in a star schema, the idea is that you have uh you have this sort of center point, the center table here called the fact table. And think of this as like the thing you're recording some event over and over again. So again, if you're like Walmart, the fact table would be every item that any anyone has ever bought at a Walmart store would be another row in this fact table. So this thing is massive. Like you know, think of like Amazon or Walmart.
9:23 Billions, if not trillions of things people bought bought over the years. And then you have in this periphery you have these tables where you have this uh for key relationship between the fact table and these these outer tables called dimension tables. And this is the additional metadata you would use to enrich the data that you have in in your fact table. So for example, somebody somebody bought something in uh bought an item there would be a foreign key reference to a product table and in the product table you have what's the name of the product you know what was the color what was the size maybe what was uh
9:56 where was it source what year was it created right so now the idea is that I have anytime I want to do a uh I want to do any analytics on my data I'm primarily always going at the fact table and now I'm just doing a join one level deep to all these sort of outer dimension dimension uh tables Right?
10:12 So it's like you're sort of simplifying what the data looks like so that at most you're only doing you know one deep join out through the dimension tables. Right? And so this was very common in the late 90s early 2000s because the analytical systems before column stores became a big thing before vectorized execution became a thing like doing analytics was super slow in a row store right and so if you reduce the number of joins that reduce how much data you uh how you're processing crunching through
10:41 course now of course your means you're duplicating data right I could have uh in my product table my dimension table I have a category name I'm going to have a bunch of say bunch of uh I don't know diapers I have different copies of diapers or different types of diapers, brands of diapers. They're all going to be in the category diaper, and I'm going to repeat diaper category name over and over again. So, I have I'm going to have duplicate data, but that's okay in some cases because I'll be able to run queries faster. The alternative is to do what's called a snowflake schema.
11:06 So, you just think of a star schema as a restricted form of a snowflake schema. And the idea is that you still have this centralized fact table where again, everything anyone's ever bought over time, but now your dimension tables can go multi multiple levels deep. I think of like the the arms of a snowflake spreading out from from the center point. Right? So now for my my product dimension table, I've uh I've factored out the the category name.
11:31 Now I have a foreign key reference to another dimension table called the category. And so if I want to get the category name and additional attributes for the category, I would do another join uh from this outward. Right? So the the difference when we want want to use one versus another, right?
11:49 So in the case of the snowflake schema, it's going to take up less storage in our system because we don't have to have this duplicate, you know, attributes over and over again, right? Uh, and often times that's a good thing because when you have duplicate data, anytime you got to update something, which doesn't happen that often OLAP system, but certainly does like if the category name changes, I got to make sure I find every single copy of where I've duplicated diapers and and change that everywhere.
12:14 And the data system can't do that for me automatically because it doesn't know that these things are linked together. You have to do this in your application code. Right? So, we haven't talked about normalization or normal forms. Uh, I don't teach that because you don't actually need to know it. Uh, just be aware that, all right, this is the one lecture. There's a thing called normal forms. It just tells you how much you should break up tables, right?
12:36 So, Andy's orders and Andy order items. Should I embed Andy's order items into the Andy's order record or should I keep them separate tables. If you break it apart, it's called normalization. If you put it inside, it's called denormalization. Sometimes you want to do it, sometimes you don't want to do it.
12:50 That's all you need to know, right? So we used to teach like a whole lecture on like function dependencies and arm aims and then you go in the real world nobody does this stuff. If you've ever written an application using like an OM think of like active record or hibernate and you you create objects and you link them together foreign keys you you basically end up in third normal form automatically without any thought right if you start to care about these things then this is where you pay a DBA or someone who knows what they're doing to figure this out but 99% of the time you won't need to know normal forms.
13:18 So going back here, this is an example of denormalization, right? So I took dimen product dimension, if I collapse it all together, that's denormalized. If I break it up and cross 12 tables, that's normalized. And then in theory, you just keep do this forever. There's like fifth normal form, sixth, like now in the theory world, nobody cares about third form, BCNF is probably good enough. But that's all, okay, this is literally all you need to know.
13:40 You might see it in the real world. Someone brings up normal forms. It just means how you split things up. Okay, so the the normalization can be good or bad depending on what the how much extra work I'm willing to do in my application to make sure my data is all in sync. And then the complexity of the query is actually going to get harder in a snowflake schema because now I'm doing like three four level levels of joins
14:03 whereas in a in a in a star schema because the dimension table can only go one level out then that's just going to be a left deep join and we know how to rip through those very very efficiently. Right?
14:13 And the you still maybe want to figure out what the join order is for the dimension tables, but you know that the the last table you always want to be joining on, right? The ride through in the pipeline is going to be the the fact table because that's going to have again trillions of records and you're just going to go very very quickly in this, right?
14:29 So most nowadays every every analytical system uh that that people are actually widely using is actually going to support both of them. It's not like you would go into the system declare that you were a star schema or snowflake schema. you just do whatever you think you want to do setting up in in in these in different ways. Um, and certainly if you don't use the snowflake schema, every every major analytical system will still do joins for you, but it might not be as efficient as setting it up in in this way.
14:53 You ever want to know why snowflake is called snowflake? Because of the snowflake schema, right? But in the early systems, you could only do you could only do star schemas, not snowflake, but like think like early 2000s. All right? All right. So that's the environment we're trying to deal with. We have these giant fact tables and we're trying to join them with dimension tables. Could be one level deep, multi-level deeps, it doesn't matter. That's what we're trying to make work efficiently.
15:15 But now we're trying to do this in a distributed environment. So basic idea is that application server comes along, sends a query and say our data is now split up across different partitions that we want to uh do a join on, right? And so we want to have the illusion that even though our data is split up across multiple resources, whe whether it's shared disk or shared nothing, it doesn't matter. We want it to be able to take the same SQL query that would run on a single node and distribute across multiple nodes and still produce the correct answer.
15:43 So the stupidest thing to do would be just to take all the data that I want to use for this query and copy it to one node then compute the join and then return the result. It would be correct but you know defeats the whole purpose of having a distributed system because now I just put everything back on a single node. And certainly if this thing maybe has the data set maybe is like pabytes and this single node only can hold a couple couple terabytes of data and only has a small amount of RAM then I'm going to easily you know run out of space quickly and not be able to finish the query. So that so we're trying to avoid this.
16:17 We're trying to distribute things across and the main thing we're going to care about is is joins and moving movement of data. Right? In in analytical world analytical workloads the queries are mostly read only. So yes, we have to update things, but it's not like last class. We've cared a lot about making sure that if I do an update on one node, it gets propagated to another node and everything's synchronized and there's a crash and make sure everything is is is all correct.
16:40 We still have to worry about that, but it's not the most critical thing, right? The thing we're trying to really avoid is having to move data around or minimize the amount of data we have to move around uh to avoid having to again uh overcome overwhelm the network or have one node become burdened.
16:59 All right. So, first we're going to talk about again high level what query execution looks like in distributed world. We'll talk a little bit about what query planning looks like in this world as well. Uh but we won't we won't go too deep into it. Uh then we'll spend a lot more time in talking about uh cloud join algorithm or sorry distributed joint algorithms and then we'll finish up doing a quick sort of preview of what cloud systems look like and a lot of things we'll talk about will be relevant both for OLAP and for uh uh OLAP and transactional systems.
17:23 But then that'll set up set us up for next class when we just pow plow through a bunch of these distributed uh OLAP systems. Okay. All right. So as I said the the because SQL is declarative because the application doesn't want have to know or have to care where the data is actually located and what what nodes are going to run it. Right? Again, we want to give the illusion that we're we're we're we are a single logical database, even though we might be broken across multiple multiple resources.
17:52 And at a high level, the way we're going to execute a distributed query plan in an analytical world is going to be basically the same as we do in a single node system, except now when we have a a pipeline breaker, we may have to send data to another node and we may have to wait for the data we need to come from from another node. And we obviously want to avoid any false false negatives or false posit false negatives where we try to maybe do a join locally and the data that we need to do that join hasn't shown up yet.
18:20 So we do a join against some hash table that's not that's incomplete and we produce incorrect results. So we need to be aware of where data is coming from in our query plan and ideally you know actually we have to wait for it to arrive at the right time so that we don't end up missing things.
18:36 So all the the stuff that we talked before how to do table scans, joins, aggregations and sorting again all of this just way more complicated in OLAP system but the core algorithms by themselves the way we want to sort of divide and conquer partition things all that is still going to be the same. So here's a rough high level architecture of what a distributed OLAP system would look like.
18:56 So, so say we have some worker nodes and it could be shared nothing or shared disc. It doesn't matter, right? We have some some nodes that can compute queries for us. And so the first step is we got to go get uh what we'll call persistent data. And again, this could either be on the local box if it's share nothing or some S3 object store if it's if it's share disk. But think of this as like the leaf nodes of the query plan that start producing the scans doing the scans over our our base tables or underlying data and start producing some results.
19:23 And then as they crunch through the query plan, they're going to produce what we'll call intermediate data. And again, this is just the data we're passing from one pipeline to the next or one operative to the next. But in a distributed environment, again, we might be accumulating this and then have to send it off to another node for further processing, right? It's not like we in when there's a single node, the pipeline ends and here's all our intermedate results from the last pipeline and the next pipeline just picks up and starts using it, right?
19:49 So there's a bit more coordination we have to do to figure out where this thing needs to go. So one way to do this and we'll see this uh in further detail later in the seme the lecture is that you could have a shuffle node uh where you're just sort of sending data to this this intermediate layer and this could either be a standalone service or it could just be the nodes themselves that could be involved in this it doesn't matter and then when the the the next phase of the pipeline the query plan starts the next worker nodes come along and they start retrieving data from the shuffle shuffle node uh the shuffle layer and then they process start processing the next part of the pipline line and you just keep doing this over and over again until some point you traverse or you you converge root of the query plan and then now you have like the final result that you would then maybe send to a single node to put it all together and return this back to the client.
20:39 So above all this what we're not really going to talk about is that there's going to be this coordinator component or scheduler if you will that is aware of what queries are running at what stage of the pipeline they're running where the data is coming from where it needs to go and it keeps track of like if a node goes down how do I replace it uh if a node's falling behind if a straggler how do how do I maybe spin up new nodes to to to run it right so there's something there's a sort of brain up above that's figuring out what's what's supposed to happen next as you run the query again. Yes.
21:08 Is this similar to map produce or I guess it's not question is this similar to map produce? That's a uh touchy subject. Uh map produce is garbage, but yes. So I would argue map produce copied this. They didn't know it though, right? So this well so the shuffle piece well I'll cover this later.
21:24 What does shuffle sound like? We already talked about this when we talked about parallel execution exchange operator, right? They didn't call it exchange, they called it shuffle. But yes, so uh without going too much details, map produce came out. So map reduce came out 2004 2005 there's a paper from Google that said hey we have this new programming paradigm we have this map function and reduce function and everyone said oh wow Google's the hot company this is clearly the way you want to do distributed uh data processing and then they they people started building clones like Hadoop and map R and Horton works uh and then all the data people basically said wait a minute wait a minute you're just reinventing we did in the 1990s in parallel databases and you're actually doing a terrible job at it uh and then all the Matt produce uh afficionados or zealots that were like, "Oh, the data people, they're all old."
22:12 This would be like Mike Stonbreaker and Dave Dwit, the people that invented all this stuff. Um, and then I was involved in this. We wrote a paper in 2009 that basically showed map produce was super slow and inferior to a a distributed OLAP system like Vertica. Uh, and then basically Matt Produce died. Not because of our paper, but it's part of it, right? And again, what what does Snowflake do? Snowflake basic so when Matt reduce came out, people were like, "Oh, yeah, this is great.
22:37 I can do you know large scale data processing fantastic uh and I don't need a very very expensive distributed database system which at the time I agree with that right there wasn't a good scalable open-source solution right and so then people realized oh writing Java functions to process data that sucks oh SQL that's a good idea let's add that back so then they had things like hive we take a SQL query and would convert the SQL query plan into map produce job and then run your map produce job for you right but even that was an in inferior implement limitation to what a a you know parallel distributed database could do.
23:09 Um, and what the Snowflake guy got right was they bypassed all the hype around map produce and said, "Oh, SQL relational model, that's a good idea. Let's build a scalable system that's going to run on the cloud and be shared disk."And then when when people, oh yeah, realized SQL people so came back around and said, "Yeah, SQL is a great idea again."Uh, and then Snowflake was right there at the right place at the right time and that's why he's worth billions, right?
23:33 Um, so yes, is this map produced? Yes. But what I'm describing here is from the 90s and map reduce borrowed from this. Don't use net reduce. Yes, this may be a silly question, but what is the point of shuffling the data? Question, what the point of shuffling the data. We'll get to this later, right? Snowflake doesn't shuffle. BigQuery does. Spark does.
23:51 And I had dinner with the Snowflake co-founder a few weeks ago when I was in Amsterdam. He's like, "Oh yeah, if I had to do it all again, they would add a shuffle."You'll see why you want to do this. It makes things a lot easier. It seems like it'd be an extra step, make you slower. It opens up additional things you couldn't do otherwise. We'll come to that later.
24:10 Okay, so I've already sort of said this, right? The category between persistent data and and intermediate data, right? Persistent data would be data that's coming from uh either like an ultimate database or like the data that's that's saying this is the primary copy of the data that we have, right? Um and typically again if you're storing these things on S3, those files are going to be immutable. like a bunch of parquet files that are sitting in S3 you can't do you can't do fine grain updates to them right so just you're just sort of reading them sucking them in and as a beginning stage of of executing query and all the results that you're generating derived from the persistent data from doing joins aggregations other things those that's called intermediate
24:51 data and you have to send me those around to the different nodes based on who needs who needs to read what so the big decision we have to make now in our distributed system is how we're actually going to process this data
25:05 And again the the distributed system knows where the data that you want to access at least in the very beginning where it's coming from and then all the animate data obviously knows where that is and where it's going. So the question is how do you start moving this data between the different the worker nodes and a high level the two approaches are well high level there only two approaches is going to be push the query to the data or pull the query pull the data to the query.
25:27 So push the query to the data means that wherever the data is residing either at another node or a you know a distributed system whatever that rather than sucking out all the data bringing it to you where you want to compute your query you send the query to the where the data is located and let it process it and some back send back the results. Right? This seems like it's kind of obvious that you want to do this because if I have one pedabyte of data and I'm doing I want to do a scan on it and I have a filter that's very selective.
25:55 It's going to throw away 99% of it. I don't want to send all that one pabyte over to the network to my node just to end up for me to throw away all of it. Right? So this is what the shared nothing systems could could do uh were very good at. uh but they have other problems as we talked about before where uh you know if you have to add new nodes you have to copy things around and that becomes expensive right
26:19 pulling the data is typically thought of how you would do this in a in a shared disc system at least in the very beginning when you're reading persistent data because I have my data on S3 I can't but you kind of can we'll talk in a second you can't really run queries on S3 so you got to read all the data out of S3 bring it to your node then do whatever computation you want on it and then after that you can then you know do the the the the push the query to the data method once you get further along with with intermediate data. Right?
26:48 The way is like the the SQL string itself is going to be maybe a couple tens of kilobytes. The data itself could be pabytes. So it's much better to be sending you know a small SQL string to go run that than a bunch of a bunch of data to process it. And the reason why I'm saying like it's the lines are can be blurred on this because Amazon S3 will actually let you run SQL on data sitting sitting in S3.
27:08 So you it's called uh S3 select. So you can basically pass along something that looks like SQL to S3 and say I have a bunch of paret files, you have my CSV files and you can basically do predicate push down and projection push down and and S3 will crunch that you know your data and only send you back what you actually need. Azure has the same thing.
27:32 Uh I don't think it does SQL, but they add Oh, it does use SQL, but you basically you can pass along in your when you do a lookup like some predicates and filters. And again, they'll they'll do additional processing for you. I have to double check this. I don't Google I don't I don't think supports this. And actually, I don't have the screenshot here. Amazon just announced like a week or two ago that they're they're not sunsetting this SQL select thing.
27:54 Uh but if you're a new if you're a new customer, you can't get it. existing customers can get it because you think about it, this is expensive, right? You're you're you're doing much of compute and they're not charging you for it. So eventually they're going to figure out how to charge you for it because again, you can do if I have a pabyte of data, I can send them a select query that then throws away almost all of it.
28:11 But they're the ones crunching through that data uh something I I assume looks something like like Redshift uh to then filter out the things you need and send it back. And I don't think they charge you extra for that. Right?
28:24 So again the lines can be a bit blurry but but when S3 for the most part object storage you think of like you're you're pulling the data to the uh to the query rather than pushing the query to the data. This is just repeat what I've already said right so say if I'm on a share nothing system or sh uh I send the query over here to the one node the one node recognizes okay well the data I need to to to process this query is locate this other node.
28:47 So I'll shove down the join request to this guy here. He does the processing locally and then sends the result back up versus what I showed in the very beginning which was a bad idea where this node would send all the data it has up to this other node and then it does the processing there. Right?
29:05 Again, we lose all the parallelism of having extra compute nodes if we shove everything to a single node because I got to copy the data all over and then do the join for my data and then do the join that for the data that got copied to me. Again, this is only two two nodes but think again you have tens of nodes. uh this would be expensive in a shared disc system.
29:22 It basically works the same way, right? The query goes here. Uh you send the the the plan fragment down of like, hey, do this join. Each of these guys can then go to share disc and just get the data that they need, right? Either the pages or the files, whatever. They get them get a local copy. They each do the join locally and then this guy sends the result back up, right?
29:46 So this is t typically you know the it's it's we're still splitting and partitioning the data up the qu and letting the the nodes process them in parallel and so even again even though this is a uh this is a shared disk architecture
30:03 this phase here I had to I had to pull the query pull the data to the query but then after I do the processing all the inate results get pushed to wherever who who needs them Next. Okay. All right. So, it's important to understand that the again we still have all the things we talked about in this distributed architecture that we we talked about earlier semester. So, there's still going to be a buffer pool that we're going to maintain at every single node so that when data comes in,
30:32 right, we store it in pages in that buffer pool and if we've run out of space, then we go ahead and spill it to disk, right? Right? We can be a little more clever with scheduling and recognize, okay, well, this is the data that's coming in and I know immediately going to going to process it as it arrives and so maybe make sure I keep it in pages that are pinned and don't let it spill to disk and like if other data comes while I'm still processing the first batch of data then that maybe it gets spilled to disk.
30:53 There's a bit more orchestration you have to do to keep the system running uh running efficiently. Uh but at a high level the bufferable stuff we talked before is still the same. But now the challenge is going to be that if I start processing data as part of a query and I produce results and maybe I haven't sent them out yet to the next node or to either the shuffle service or another node the system might crash like my my single node might crash
31:22 and in that case that portion of the data that would being used to process the query that's now missing from the final result set. So traditionally before the map reduce stuff came came along traditionally what the way a data system would handle this it would say okay well if one node goes down during query execution it's rare and if it happens we'll just kill the query and restart it
31:46 right but again now if you think if if in like the map produced world they were running on thousands of machines most databases aren't running that on that side most databas are running on tens of machines but still the likelihood that something could crash would could happen And if my query is running for days, I don't want if one node goes down, I lose all that work. So this idea that we need fault tolerance for the intermittent results part of our query so that if one of our compute nodes or something goes down during execution, we don't have to throw everything away, right?
32:18 Another thing about this is like I don't recommend this, but you could run your data system on spot instances on Amazon. Everyone know what spot instance is? Right? Amazon basically says it's non-reserved instance. they can take it away from you many time but we'll charge you less for that you could build your data system on that some systems do uh and then they have this this notion of fault tolerance so that like as I'm running I'm paying less for than than a reserve EC2 instance and you know what if the thing gets crashed or taken taken away from me no big deal I don't have to restart the entire query I just pick up another one and keep running right
32:51 so the idea is that we want to be able to have a sort of a snapshot of the end results while we're running so that if one of the nodes goes down other nodes can can pick up the pieces and and keep running from it, right? All right. Well, how are we going to do this? If I just said that we have a buff manager, we're getting, you know, processing whatever data we have, we're producing intermedate results, and then maybe we're storing that on our local disk.
33:10 If that node goes down, we can't get it back, right? Because the the disk is attached to uh that node. So, that means we need storage area. We need discs that aren't attached to any one compute node. What is that? S3. Exactly right.
33:28 It's the shared disc stuff of we talked about before. So now what I'm saying here is when we execute a query, right, this guy sends, hey, do the join on RNS. This guy can go get some uh you know do whatever processing you want either from local data or go goes to S3 and get it. But now instead of sending the result directly to the node above, it just writes it out to S3 or some distributed caching layer that that is disconnected from this.
33:52 So that when this guy dies, there's again there's some schedule coordinator above that's keeping track of what nodes are part of running my query. It would recognize I don't get a heartbeat back from this guy. It's dead. So instead of restarting the query uh res excuse the query plan, I could go look and say, well, I know I told it to go write out its join result to shared shared storage at this location. Let me go see if it actually made it there.
34:17 If yes, then I can just pull that and and resume execution. maybe spin up another node because it's stateless to let it keep running that part of the query plan, but I don't have to stop everything and start all over again. Right?
34:30 This is something the map produce guys did, the Hadoop guys did that the earlier distributed data system did not do, but this is pretty common now in in relational distributed data database systems. All right. So for query planning again all the same optimizations that we talked about before like predicate push down projection push down the optimal joint learning all that still applicable here right what's different now is that the instead of maybe counting dis IO's as we did when join algorithms maybe got to care about network cost like where's the data located where does it need to go how's it partitioned or not partitioned is it replicated or not
35:08 right we can account for that in in our cost models to make better decisions Uh we'll cover more of this again if you take the uh the special topics class next semester. But in some systems they'll do they'll do sort of single node planning first like to solve that problem and then they would then do a second phase immediately after to go then hack up your single node query plan and make it distributed.
35:30 This is roughly how Microsoft does that in in in SQL Server at least at least from what what they tell us, right? I'm not saying it's right or wrong. I'm just saying that that's what they do. uh some systems and then we talked about where uh instead of having like hard-coded values for like oh here's what a network transfer cost is when the system boots up they do some micro benchmarks really quickly to say how much how fast can I send data between different nodes in my cluster
35:53 and they'll use that for their cost model estimates right so there's a bunch of different tricks you can wait you can handle this yes so in the previous slide if the node failed before you can send a result question is in my example here if the bottom node failed before wrote wrote the data S3. What happened?
36:10 You basically you the schedule coordinator could recognize okay I asked you to process this d this worker to process this data it didn't complete it let me assign that task to another node or a new node and they can spin up new ones huh what to
36:27 so we as like in the very beginning right for this query here we ask the top guy asked the bottom guy hey do a join RNS right and if that if it crashes before I write that result there right you think of like I write the result then I tell the schedule coordinator hey by the I finished. Here's where to go find it. Right? And that notifies it that it's completed. If you don't see that notification, then you know it didn't actually complete it.
36:48 You can then assign it to another, you know, node. Also, it doesn't have to crash, right? What if whatever say this is running EC2, you ran on a bad box where they there's another tenant on the same machine as you and they're like Bitcoin mining or doing something shady, right? That's going to slow you down because they overprovision the hardware. And so say this guy's just it's just a slower box for whatever reason. So it looks like it's dead even though it is producing results.
37:10 So if by the time everyone else has completed this computation and this guy's falling behind, you can say, "Okay, they didn't complete it. Let me sign to somebody else."There was a I can't say who. Um there was a Davis company or there was a there was a web app company a while ago and they were running my SQL and every time they had to spin up a new MySQL instance on EC2, they would spin up three instances, run micro benchmarks on all of them, figure out which one was the fastest one and killed the other two, right?
37:36 And they were just doing this constantly to figure out what what nodes were were not having noisy neighbors. Okay. Uh so we we tal again we already talked about this how the query plan and we have to care about all the things we talked about before but now the question is what are we actually going to send between different nodes. I guess the spoiler is going to be everyone's going to do the first one sending these physical operators.
37:57 Uh but there some systems can do can do send SQL. So the idea is a memory we take a SQL query we generate an a then we generate a logical plan from a logical plan we generate a physical plan that says here's the actual operators that I want to execute in this order and on on this this portions of the data right so when a query shows up there'll be sort of this centralized coordinator or some some some part of the system say do all the query planning for you and it's trying to build a global optimal plan that is going to distribute across all the different different nodes and what it sends is actually going to be again a physical plan it could be something like substrate
38:32 uh which we can cover next class um different ways to represent a query the standardized way to represent a query plan but it's essentially going to be the physical plan that we want to send to the different nodes and so then now when a compute node gets that physical plan it just says okay here's the task I got to do let me just go execute it and produce a result
38:49 it's rare but there are two systems that do send instead of sending the physical plan after they generate the the an essential location the the physical plan from a SQL query they then break that up into subtasks uh thing like breaking up portions of the of the query plan and then they convert it back into SQL that they then send to the different nodes who then can then take that SQL and do all the parsing, planning and optimization that we talked about before, right?
39:13 The idea here is that you you have a you would assume that the your global view of what the data looks like in a centralized location is never going to be exactly up to date and therefore the local nodes could be in a better position to decide what data actually looks like and can make their own local optimizations. Right?
39:34 So single store does this uh and vitess does this. The test is uh it's a it's a distributed middleware for uh that YouTube invented for my SQL. So like a sharded version of my SQL. It's not really for analytics just this is what they the way they do it. So they have your query shows up into this my SQL front end. They generate the physical plan and then they send them the the query plan fragment to another node. They convert it back into SQL.
39:58 So then my SQL on the next node can then just do all the processing as if it was a local query. Again it just looks like this. Again I have a SQL query. I can then generate sub plans to say all right for this node it has values from ID 1 to 100. So the query plan they're going to be getting is just with a wear clause with I ID 1 to 100 right and then now all these different nodes can just take this take this query plan and do a local plan and maybe the case that this chooses a different joint algorithm than this one based on whatever data it has and again that's okay because logically it'll still produce the same result. Yes, sorry.
40:38 For the second approach, do we um generate a optimized plan first, convert it back to SQL, then send this translated SQL to the node? Why don't we just directly send like why is the global optimization steps to necessary given that we are also optimizing locally? Her question is why even bother doing the global optimization? You kind of have to do that in order to figure out like what portion of the query needs execute on different nodes.
41:02 So my example here again I'm just doing a join on RNS, right? I mean, you could just send this you could just send the top part down here and just say, "Oh, only process the data you have locally. Don't don't because you don't don't worry about communicating with everybody else."It's basically equivalent to the same thing. So maybe I should pick a better example, but like you could imagine where something like I don't want to I want
41:27 uh I maybe have additional information in my query plan about where data is going to go like through multiple stages instead of this is just like everyone produces the same result or excuse different part of the query and then union result at the end. I may have multiple stages where I still want to have I still want to have a global plan that I have to optimize first.
41:46 Right? In this case here, it's joint R and S. But say there was a third table of join R, R, S, and T. I still have to know at a global level what order should I, you know, should be the first level join and should be the second level join.
42:01 And the idea is get the like the centralized node may not have up-to-date statistics, but again in a in a lakehouse system like a data bricks, we'll talk about in a second, uh there's barely any statistics anyway. So this doesn't really make any sense. Okay, so most of the time on a single node system, the most expensive part of a query plan is going to be the join.
42:25 We've said this multiple times, the joint is always going to be the most expensive thing, right? Depending on if your your disc is really slow uh and you have a lot of memory, then maybe the join is not the most expensive thing. Just getting things off the disk is expensive. But in in practice, in general, the join is going to be maybe 50 to 60% of the entire time of a query. Right.
42:44 So the question is going to be now how are we going to do this in a distributed environment where the the data we need to process may not be aligned correctly or aligned based on what our join keys are in in our query.
42:58 So we saw what we did before right I could just take all my data across different nodes and plop it into a single node. But again if if that's pabytes of data that's not practical that's not feasible and I have to pay the cost of transferring data over the network. We're also assuming we're running in the same data center here. That's a fine assumption for this class. But again, it could be the case you're trying to join data at different data centers.
43:17 Now you're paying Amazon or Google, whoever your cloud provider is to to send data out and into the different data centers, and that's going to be expensive, right? So again, we want to minimize the amount of data we have to send between the different nodes in order to process joins. So say we have a really simple example. we join RNS and there'll be four different scenarios we have to account for of how RNS could be stored on in in our database system and again it doesn't matter whether it's a shared disc system or whether it's a shared nothing system
43:48 the end of the day there's there's there's going to be data and it may or may not be partitioned it may not may or may not be replicated at the different nodes that we want to use to compute a join
43:57 and again the goal is we want to produce the same result as if we were running on a single node the the sort order could be different. That's okay. You depending whether there's an order by clause or not, right? The sort order could be different whether it's a single node or or multiode, but you get logically the the the result has to be the same. All right? All right.
44:17 So the first scenario is the sort of the most simple case where the we have a table that's R that's partitioned on the ID and it just so happens that the the I R ID is what our join query is based on. So the join clause says do a join on on on the ID field and then there's a table S that's smaller that's replicated at every single node. Right? So think of again ID is the partition key and then this metadata information here is just what range of values do I have for this this ID here and then S is just replicated ever.
44:53 So again think of this as like the R is like the fact table where that you have this huge thing and that's partitioned based on like the the sale ID or something and then the S is like one of those smaller dimension tables like like zip code. There's only 40,000 zip codes in the US and that's just replicated on every single note. So this is great when we want to do a join because now we just do a local join at every single node with whatever data we have in R and the replicated copy of S
45:20 and then we just ship the result from one of those nodes to another node union those two things together which is cheap to do just pending you know one block of data versus another block of data give that to the the client or whatever the next stage is in our query plan and we're done right so that's easy that's nice this is like the ideal scenario if if your data looks like
45:42 Another scenario might be uh where again it's you're still partitioned on the uh in the join key for R but you actually are also partitioned on the join key for S as well. And again just like in before with the replicated data in the first scenario. This is great because I don't have to send data between the different nodes because all the information I need to join R and S at each node is local to me.
46:04 I just do the local hash join sort join nested loop join for crazy like do all that locally just as before produce my result ship it over ship just the the join result over concatenate the two result sets together and I produce my final value right
46:22 so this is the second best scenario because again I have I mean in some ways it's it's indifferent to the what the first scenario whether it's replicated or partition not but like I'm partitioning exactly the keys I need to do my joint
46:35 So I don't have to no node has to go sniffing around and talk to other nodes to figure out am I missing something I need to my join to avoid any false negatives. So this is nice when it happens, but this is this this can be rare, right? Because you may store a bunch of data in a bunch of parquet files in S3 and you didn't do actually do any partitioning because you don't know what queries you're going to execute later on and then some guy comes along a year later and starts doing some join that something you didn't even consider before and now your data is not nicely aligned with the how things are partitioned or h how the join the join key in your query wants it.
47:12 So the third scenario is when say the first table is partitioned on the ID we want to join on but the second table is partition on a different attribute like a value attribute. Right? So now again I can't do a local join on every single node because I may I may get false negatives. I may be looking for say ID ID 33 and I look here and it's not in my copy or my portion my partition of of the S table. is actually over here but I don't know about it.
47:44 So what I need to do is I need to basically repartition or sorry I'm gonna send my portion that I have or every is going to send its portion of S that it has to every other node to put us back into the first scenario where S is now replicated at every single node. So the first guy here just takes his his copy of S that he has sends it over the wire and and appends it here.
48:06 And this guy does the same thing. Sends his copy of S over here. Append it. And now again, now I'm in the first scenario where S is completely replicated in its entirety at every single node. And then I do just do my local join that we did before. Produce the correct result. Ship that result set over the wire to one of the nodes. I pen it together and I'm done. So sometimes you'll see this called as a broadcast join.
48:27 So they'll say say it's a broadcast join. It doesn't really tell you what join out of them they're going to use for the local join, right? It's just the idea that you're broadcasting your copy of some table or data set to all the other nodes so that you can again get a complete replicated copy of that data at your node. So you say I'm doing a broadcast join or broadcast hash join broadcast service join. The broadcast piece is that extra step.
48:52 And in this case here you have to wait until you get the entire copy or all the data you want for S before you start processing the join. like you could still to build the hashtape if you wanted to on R
49:03 uh and then do the do do the probe on it if you wanted to but you know you you could do still do other work while you're waiting for the copy to S but you can't complete the join itself until you get all all the data. Yes.
49:15 So in this case do we assume that S is a smaller table? His question is do we assume that S is a smaller table? Absolutely yes. Because otherwise the problem we had beginning where you're sending a pabyte of data as a one thing node and it gets overwhelmed. Okay. Okay, so that so this is the third scenario.
49:32 The fourth one is the worst case. This is where now both R and S are not partitioned on my join key and they're not replicated and they're too big to put on a single node. So I I got to do something different. So this this this is sometimes called a shuffle join. And the same way you say you're doing a broadcast join, but it's really like a broadcast hash join. You say you're doing shuffle join, you're probably doing a a shuffle hash join.
49:56 And the basic idea is you're just going to repartition the data so that it aligns with what you want your joint key to be or what your joint key is based on. So I take my copy of R that I have basically hash the the ID field because again assume I have all the attributes for this this data my node and then just send whatever the you know this range of data to this guy over here.
50:19 I'm doing range partitioning instead of hash partitioning but the ID is the same and this guy sends his portion of the data for for R over here. Then I do the same thing for S. And then now we're back in the second scenario
50:30 where all the data I needed uh to do the join RNS is nicely partitioned exactly on the join key and all the data I need is is there and I just produce my join result and append it together. So he was asking why why would you want to do a shuffle?
50:45 It seems like an extra step. this is an example what you want to do with it, right? Allows you to be able to redistribute the data based on uh to partition based on a key that you know you're going to need for the next stage in a query plan. So as I'm producing my table scan, if it's un if it's not um if it's not partitioned on my join key, then I can send I could do that intermediate shuffle step then to then partition on that join key and then send it out to who needs it. Yes.
51:13 What if I join on multiple attributes? The question is what if you join on multiple attributes I mean it's just you just it's partition multiple attributes right it's no different like again say this is range partitioning say it's hash partitioning it's the same thing when we did multiattribute join or multiattribute hashing in a hash table locally you just concatenate the bytes hash it and then it tells you where you need to go again all the same primitives and techniques we used entire semester we're now just throwing in a distributed environment but it's it's the Okay. Yes.
51:50 I don't know if we talked about this, but if each node needs an index, do they index the entire data or just the stuff they have? This question is if if each node needs an index, would they build an index on the entire data set or they build on on the local one? So, in the OAT world, you typically don't build indexes, right? Because you're not doing like a B+ really great for like point query lookups.
52:12 I'm not I'm not typically doing that. I'm trying to rip through everything uh for OAS. So OA systems, yeah, you typically don't build an index and if you did, you you would build it to the local to the data set. Um but you would do like zone maps would be like uh like think like minax values for columns and then you put that in like a centralized location and use that to do early pruning like oh I I know I don't need to read this file because it's not going to have data that I need.
52:38 Um in the transactional world typically you do it in most systems the the index is not rep is not global. Uh and so that means that if you do a lookup on something where like a on a on a key that isn't what the partition key is you have to do a broadcast of that. There was one system called cluster that came out of AOL that got bought by
53:04 oh but got bought by Mariab uh and then it got renamed to cluster or marb expand with an X because whatever uh and then I think they killed that off but I think that one had a you could build a global index. So if your query showed up and it wasn't on the partition key you'd hit that global index first and they had a way to keep that synchronized and then they could figure out what node you need need to go to next.
53:25 But most distributed OTP systems don't do that. So yeah, if you do a but like that world is different. Again, think of like going back to this like here's Andy's orders, Andy's order items. Like you're given Andy's account ID. So you so you you would know how to find the node, the one node that has that data. This is like trying to scan everything, not looking for Andy's records, looking for everyone's records. And therefore, you don't need to worry about the, you know, finding an index to find individual thing.
53:52 But I still need to know how to do the join. And then you may say, okay, well, building this hash table is expensive. All right, if I'm going to see the same query over and over again, instead of having to do this shuffle over and over again, could I just cache the inate result so that when I come along and do the query again, I don't have to, you know, send data around. Some systems do that, some systems don't, right? We saw this on fireball.
54:14 Firebolt talked about how they trying to build a cache service so that they can have you reuse in results. That's what they're talking about. They're trying to avoid this shuffle step because in their world they're paying to go read data from Amazon, right? So they want to they they you're going to pay the same amount for the query whether or not you read from Amazon or not. So they can avoid talking to Amazon. They save money when they run your query. All right.
54:40 So the last one is is is is an optimization we can do when we in this environment. Um we don't really talk about semi joins this this semester. just just be aware that that exists. Um I don't the SQL standard doesn't I don't SQL standard doesn't doesn't say there's a semi- join. Some systems you can declare a semi- join but again this is something that you should the system should just do automatically for you. And the basic idea is that if you can recognize that you don't need the entire data set or all the columns or attributes to do a join, then instead of sending entire copies of of of tupils
55:15 over the wire to another node, you just send a compact form of it either doing projection push down only send the bare bare minimum data you need to do the join or in some cases you can send a filter data structure like a bloom filter. So the say the idea say that my fact table is this massive thing and then I have this dimension table that's really small. So I want to do a join between these two guys.
55:36 But before I just before I maybe copy all the data back and forth, I'll first do a local projection and filter on the dimension table to produce a uh a minimal form of it. Send that over the wire. Now I I do my my join locally on that. But then what I'm going to produce out is just again the bare minimum I need to complete the rest of the query. I then send that back over to the to the first node who can then do its own local join. I think it's just like projection push down sending I'm sending the if I'm doing join on the ID field.
56:14 I just send that ID field and not anything else. And then when I come back and get the result after not having to send all the of it all the data I can then do you know bring along the data I actually need to produce the final result
56:30 sort of seems obvious but it's called a semi join and again it doesn't have to be like you know a list of ids you could actually build a bloom filter and again we talked about before like the same way like oh people say I'm doing a shuffle join or doing a proccast join but it's really a broadcast hash join or a shuffle hash join people say you can do a bloom join but it's going to be a bloom hash join. Right?
56:49 There's there's doing this extra bloom filter step and that's that's called a semi- join. And this works because again, if you look at my projection output, I'm doing for the fact table, I'm only getting the price and then I'm getting all the data from the measure table. If I had factstar, I'm getting all the columns from the the the fact table, then I couldn't really do this optimization, right?
57:11 Or at least sending the minimal back data back in both directions. Okay, so we talked a little about the shuffle phase, but you know, for the most part, we going to assuming that nodes can talk directly to other nodes, right? That means that when I have my query plan, I produce, you know, that I generate at the very beginning, I'm going to have be aware of how I'm going to stitch the path of data from coming from from one operator on one node to to another node, right?
57:39 And this means that I'm making big assumptions of what my data actually looks like. And again, if I'm reading a bunch of files on S3 uh that were generated by some outside external service or I download a bunch of data and I dumped in S3 and now I'm trying to crunch through it, I may have no have very little to no information about what that data actually looks like and my selectivity estimates could be way off and I could have either too many nodes or too few nodes to process the data that I need.
58:09 So the I what I want to be able to potentially do is have a intermediate stage that I have in my in my query plan where I can reassess what data actually looks like and then readjust my query plan and my resources based on what the data I'm actually seeing not what I what I predicted when I first ran the query. And so this is one of the advantages of the shuffle phase that that we were talking about.
58:34 So in addition to um the examples we saw before a redistributing thing when I have to do a join but you can think of like a shuffle phase this intermediate step is like a pipeline breaker where I look at all the data I just produced from from this pipeline below me and then I can decide how how to proceed going going forward.
58:55 So this is this comes out of the map produced world that he brought up uh and this is how actually Google BigQuery is implemented. Spark does the same thing. Spark you know obvious spark was originally built as a better version of Hadoop or map produce uh and in some ways it is uh but you know one of the things they added they haven't is this explicit shovel step in some systems like in SQL server
59:20 when we looked at the exchange operator that was like it can decide whether it wants to use it or not but again think of like bigquery and spark they're always doing the shuffle and that opens up opportunities again to reassess what the data looks like and and adjust things as you go along. The shuffle phase is so important.
59:36 Actually in BigQuery, this is semi-public, but they they actually have a dedicated hardware service to do inmemory shuffle. So they built custom hardware to do the shuffle as fast as possible because this is actually where you're going to be spending most of your time. So who cares how fast maybe your your your sorting algorithm is? The end of the day, the shuffle is going to be most expensive thing. So they try to make that run as fast as possible. There's also standalone services you can get that are completely separate uh shuffle services.
1:00:04 Saliborn came out of Tencent. Uh Unifl came out of another Chinese company. But think of this like like when you run Spark, Spark does its own shuffle, but it's just inmemory uh in memory local storage at every single node. You can then hook up Celiborn to it so that it can then say, "Okay, I'm doing a shuffle. Here's a standalone service here that I I'll communicate."It's basically a key value interface to say, "Here's the data, this key, and here's some value for it."
1:00:28 And then it knows how to organize that and then have have you know nodes pull that data off as they need. So this basically simplified version that we saw before. So you have a bunch of workers in the first stage. They're processing some kind of data. I think of like a a producer consumer model and then now they're producing results. So they then send out to the shuffle nodes and then say this thing gets too too too big or so the data gets too big and you try to keep keep as much as memory as you can as possible.
1:00:55 Again, it's another bufferable manager that we talked about before. But if I run out of space, I can then spill to a shared disk. And then now at the next stage when they when they want to go get data to say, hey, the data you need is also in this shared disk as well.
1:01:09 But notice here at the first stage here, I had four processing nodes, but in the second stage, I had three. So I can scale up and scale down the number of workers I have as I'm seeing the the data arrive.
1:01:21 Or if you notice that maybe one key is super hot, you can then do that recursive partitioning we saw when we saw grace hash joins to further subdivide it across more more nodes instead of having one node process you know a really large data set for the most you know some popular key.
1:01:35 So there's a bunch again there's a bunch of optimizations you can do uh when you have the ability to basically take a pause while your query's running and say okay how how should I change things based on the data that I'm seeing
1:01:47 and as I said this is basically the same exchange operator we talked about before right it's basically this bottom one here with SQL server called repartition distributed OS systems called shuffle okay so that's sort of a crash course everything you need to know in for OLAP systems obviously again we can go more details about certain topics and we'll cover some of some of these additional things uh next semester of how or sorry next class I will put it all together
1:02:15 into the big query and other systems but like the end of the day the join and the shuffle are the most expensive things you still need to get you still need the join to write uh but all the classical stuff or the single stuff we talked about before all that still applies here
1:02:30 all right so let me give me a quick brain dump in the last 10 minutes of what a cloud system looks like when someone says Hey, I have a cloud database. What does that actually mean? So, in the last 10 years, there's been a proliferation of we'll call database as a service, DBAS. And this is where there's some cloud vendor that's going to offer you a a managed database system.
1:02:48 So instead of you go, you know, spinning up an EC2 instance, going to Postgress's website, downloading the the binary and and running it yourself, you basically give them your credit card and then they'll give you a they'll give you an address or, you know, host name, a port number you can then connect to, and they manage the data system for you. So they handle all the backups, they handle all the recovery, they handle all all those things, right?
1:03:14 And as I said before, the the initial distributed systems in the space were were very much a share nothing approach, if they were distributed at all. Um, but now with like S3 select and things like that, the the lines get blurred.
1:03:30 So the very first cloud database systems, what we'll call a manage system where they basically took an off-the-shelf open- source system or single node system and just slapped it on an EC2 instance for you and again manage everything under the covers. the core architecture you of the system you're running on was the same as if you were running locally on prem versus running in their cloud service right
1:03:52 like Amazon is famous for this Amazon took Postgress made it made it slapped it up inside of E2 instance and they charge you double for it if as if you're running yourself right they they charge double for the storage uh when you run through their RDS service versus you running Postgress yourself right
1:04:08 you know they do you know somebody's like you don't have to manage it yourself they manage it for you is that worth double debatable, right? And then what snowflake came along with uh was sort of the new breed of system that were cloudnative. And this is where the architecture was originally designed from this onset to run in a cloud environment typically with a dist disagregated storage comput and storage. You have you know your compute layer and you have your S3 storage layer or EBS or something like that right um but again the idea is that the system is not meant to run on prem.
1:04:38 It's meant to run explicitly in in the cloud and systems like Snowflake can run I think they run on GCP, Amazon, I think they run in Azure as well, right? So they have all the control mechanisms they need to communicate with whatever the cloud vendor that they're running on and they hide all that from from you, right? You can't run you can't download Snowflake and run them on prem. It's only runs at the cloud service. BigQuery is the same way.
1:05:03 So then in the last five years there was this sort of new buzz word that's come along with serverless databases which seems kind of weird right because we just spent the entire semester talking about how the data system is going to talk to the server that's running on and then now people are selling hey we have a database that is serverless what does that actually mean and so the basic idea here is that instead of having a
1:05:24 dedicated piece of hardware like a machine that's going to run always running your database system the idea is that it's they're going to evict your database instance from the hardware when it becomes idle. And the various systems have v various uh thresholds decide when it's idle or not. And whether they're killing the VM or just turning off your portion of of a of a sliver of VM depends on the implementation. The basic idea is that the application server is talking to some node, right?
1:05:52 And let's say that the guy using it goes to the bathroom or falls asleep, whatever. The application is not sending any more queries. Well, again, if you're running on dedicated hardware, like you spin up EC2 instance yourself, you're that thing's just sitting there waiting for the next query that's never going to come and you're just paying for that. You're paying for the compute, you're paying for the storage, you're paying for the memory.
1:06:14 So, with the serverless architecture, again, if we disconnect the storage from the compute, when the application, you know, sends queries, everything just looks like before, right? This thing is has a buffer pool. It's paging in uh from shared disc back and forth. All that's fine. Then when your application goes to sleep or doesn't do anything, then they will go ahead and evict the contents of your buffer pool out into the shared disk.
1:06:38 Think of like the checkpoints we talked about before. It is a checkpoint of all the dirty pages, flush that out to disk. But then it's actually also going to write the the the contents of the page table itself, like what pages do I have in my page table. Like think about in bus hub that's that's an ephemeral data structure. It's some hash table you build in memory. when you kill bus stop and come back the page table contents are all gone right the pages ideally would been flush to disk but what pages were even were in there like that metadata that's gone as well so they write try to write out all the the the internal state of the system
1:07:10 out to disk then they go ahead and kill your instance and again whether this is running on a dedicated node or there's multiple n multiple data instances running on the same node they just turn turn your sliver off it doesn't matter you're still paying for the storage in this in this world because if you don't pay Amazon in money for what you're storing S3, they're going to they're going to they're going to delete it, right? They'll be happy to do that.
1:07:30 So, you still have to pay for that, but not paying for that compute that was idle from before. And then now when the application then comes back and executes a new query, there'll be a little slow start, a slow pause where you're fetching all the things you had in your buffer pool back into memory from a shared disk, and you're trying to sort of basically put the the data back in the exact same state it was before. like you're kind of almost like freezing the VM but a little more fine grained.
1:07:54 Uh and so you repopulate the buffer pool. Uh typically you get the first page you need for the query and then in the background you go fetch the rest rest in uh but again now this environment you didn't have to uh you didn't have to keep this server around for a long time.
1:08:12 So there's a bunch of data systems that are uh in the space. uh Neon is was one who's given a talk with us before like they basically took Postgress the storage layer ripped that part out and have now this this distributed page server that's doing basically
1:08:28 uh sort of this step here for you right pages out here and then when you get when you get evicted because you go to sleep it writes all that out as well and it can bring it all back for you fawn is another one that's in the space SQL server and Amazon have their own very various versions of of serverless data systems Right. So this is what it means. There's still a server. It's just the idea is that it isn't dedicated for your for your application who's not running queries, you know, for some period of time.
1:08:53 And obviously again this this will be per a query. This is more expensive. But again, if your your if your application is like sending one query a week, then this is the way to go. This is way cheaper. All right.
1:09:08 So to segue into what we'll talk about next class, I want to briefly talk about data links. And so you can sort of think a data lake is is the successor to what snowflake's architecture was like. Snowflake was a
1:09:21 disagregated storage compute and and storage and the the DS was still managing all the storage layer even though they weren't running the actual the actual system itself like it was Amazon S3 was in charge of actually storing the bits and bytes in your database but Snowflake was still keeping track of what they were actually putting in S3 for you, right? Oops. Sorry. And so typically what they would store would be a proprietary format, right?
1:09:46 So say I I would have to call create table. It goes through the compute node of the data system and the compute node is responsible for setting up some catalog that it's going to store in S3 and inserting the data that would then put into uh to to the database. Here again think like in busub busub has a proprietary format in the page table you created like the the actual bite sequences with the pages themselves that is specific to busub.
1:10:10 No other data system can can actually read that. Same way bustop can't read SQL lights data, right? Can't or my SQL's data, Postgress's data, right? All of these systems have their own proprietary things. And so now again when I have to do a query, the computer node knows to go in the catalog and go knows how to find the data that it needs. So in this model here where you all the data has to that you want to put in the database has to go through the data system itself.
1:10:34 This is called manage storage, right? The data system is managing the contents of what's in the the shared storage here. So what a data lake is basically saying that all right well not all my applications wants to write data going through the data system right I have some Python job some lambda function or whatever that it's just generating bunch of S3 file or sorry bunch of paret files or JSON files or CSVs whatever like they're all just inserting directly into
1:11:04 S3 itself now you have to have some governance of who's allowed to store what and when right that that all doesn't go away but in I don't have to go through the data system to put data into my database, my my data lake, if you will. And then now when a query comes along, you still need some kind of catalog to say here's what data you files you have,
1:11:26 but this is not going to be exactly in sync as it was in a managed service where or managed storage where as as I was doing the insert, I know what was being inserted. So I could keep track of where it was and what actually the content was. I could build whatever zone maps or bloom filters I want as the data is coming in. This is like the catalog is saying, "Hey, here's a bunch of files.
1:11:43 I don't know actually what's in them, but if you want to go process your query on it, go ahead and, you know, go for it."So, this is basically what a data lake is. And then what data bricks is branded as a lakehouse is again is is is this layer up above where you're letting anybody else store a bunch of crap in in your S3 storage or whatever blob storage you're using. Uh but then this thing here is managed by uh by data bricks and they know how to keep track of maybe what files you have simple you know public uh simple notification to this catalog and say here's my files here's they're these tables and then this thing knows how to read read a bunch of data off of these guys and produce results.
1:12:23 So this is this is basically what a lakehouse or data lakeink is. just the idea that the the data system itself the data storage itself the contents of it are not always being generated by the database system. So, so Snowflake basically had us uh pivot not pivot away they still do by default you still get the manage managed storage but they can support this as well. Uh Red Shift was originally a shared nothing system and then it became a shared disc system and then it became a eventually to a sort of lakehouse system that could read bunch S3 files as well.
1:12:55 They're always a little bit more farther behind than than data bricks and snowflake. Yes. Wouldn't it be cheaper to do compute if when the data is coming in you just store it structured as in like the transform while it's coming in rather than always storing it unstructured and always doing your analytics on the unstructured? Question. Wouldn't it be cheaper to to always do the computation to put it into a a unstructured data into a structured form as the data arrives than have processing later on?
1:13:21 Well, one is it you may not know who's actually generating this data for you. May maybe some data scientist that doesn't know doesn't doesn't have permissions maybe go through the database and just chucking S3 files there and trying to make sense of it later on. Furthermore, it's like the partitioning problem we talked before. I may store a bunch of data and try to figure out try to guess what queries how queries are going to want to use it in the future.
1:13:43 So therefore I I did my transformation or partitioned it at the beginning but a year later two years later what if like the the the demands of the application are completely different and now all the assumptions of how things got transformed were wrong. That's very common.
1:13:57 So the idea is like this storage is cheap, right? Relative to something EBS or direct attached storage. S3 is super cheap. Uh so I'll just let people whatever they want on it and then we'll just clean up the mess later on. I'm not saying it's right or wrong. Depends on again what the organization's protocols are, what the demands of the application. Sometimes it makes sense, right? because it lowers the threshold to store data in like if if I have if I have this new data set I'm generating but you tell me before I can put it into S3 I got to do a bunch of stuff to transform into a form for like some guy I don't know who's going to need to process it later that's that I'm just going to keep on my local laptop screw you right but you just kind of let anybody throw this data in your data lake or data swamp it's called then you people come come try to make sense of it later on and the catalog is sort of what they try to keep track of like here's what actually putting in so like we'll talk about in a second so things like unity and iceberg and uh and h catalog.
1:14:56 The idea is that before I put anything in here, I'll keep track of what I'm putting in up here. So I could maybe declare a schema so people can at least find what data actually have and then do whatever transformation they want on it later on to put in the form that they need. Yes.
1:15:10 How is concurrency control handled when the underlying data can be modified not by database management system but by others? He question is how is control man how is control handled when the underlying data can be modified by multiple people. So again S3 is append only I can't make implex updates to any files out of an S3 right so I can't once the files in there I can append to it but I can't modify I can't delete sorry I can't overwrite what's already in there I can delete pie p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p p pieces of it and then you have other protocols to handle that but I can't overwrite something.
1:15:41 So at the file level you don't worry about concurrency tool at the physical level at the logical level right these are just saying logically I have a table and it's match a bunch of files and now I want to do updates and deletes on them how do I handle that two more slides but this is what iceberg does right Iceberg gives you a a basically a CRUD interface you can do updates and deletes over logical data sets and underneath the covers and knows how to map that back into uh files on S3. All right.
1:16:15 So, I want to maybe just quickly go through this bunch of these things and we we can punt this next class. But one of the big trends we're seeing in in data systems is that no longer do you have a company just build a complete monolithic architecture where they do everything right where they they build the the the storage layer, the compute engine, the query optimizer, the user interface, all of that. that there's been this trend what the whole seminar series this semester has been about of these reusable components or building blocks where you can mash a bunch of things together and build a a distributed OLA system or distributed data system in a short amount of time, right?
1:16:51 It's not easy to put these things together and we saw one uh speaker this semester basically started with something like data fusion realized it didn't work for what they wanted to do and now they're they're writing everything from scratch themselves. Um but that that is sort of the dream that the idea is that that in a sort of a
1:17:07 open source data data ecosystem or data science ecosystem that you could put together these different components uh and be able to reuse and take advantage of engineering that other people have already done to get your new system up and running your new application up running and pretty quickly. So the catalog stuff is super interesting even though it seems like an unsexy thing like it's metadata about your data that doesn't sound super exciting but there's a lot of activity in the space because everyone needs this.
1:17:32 So age catalog came out of hive. Hive was the thing I said in the beginning where it was took SQL queries and converted to map reduce jobs and they realized oh yeah if I have a bunch of files on HDFS distributed storage I need to keep track where they are. They built H catalog the the cloud vendors have their own thing. Data Bricks has their own thing. And then Apache iceberg is probably the hottest one right now.
1:17:52 Who heard Who here has heard of iceberg? Data Bricks has heard of them. They paid a billion dollars for them for 40 people in iceberg this year. Another cloud vendor, I can't say who, was trying to buy them. Data Bricks caught Word and came in with a billion dollars and and scooped it out from them, right? Uh yeah.
1:18:15 So like doesn't seem like a catalog be super important. I mean iceberg does other things. Iceberg gives you that insert update delete interface on top of your files that you then store on S3. It's basically you do insert update writes I think it writes everything as JSON and the background they do compaction and convert it to parquet and then write that out to S3. But now you have a sort of unified interface of how you actually want to create your data.
1:18:34 So catalog seems like a super boring but it's a very important topic. Query optimizers we've already talked about. Uh I'll just say again next semester we're basically cover this in topic. Um there isn't a lot of activity in this space because it's super hard to do and so we're trying to do this ourselves here at CMU and we're not it's going okay. Um but Apache Calite is probably the most famous one in the space. Uh and this is basically again think of like it's a standalone service that does all the transformation rules and optimizations that we talked about before but it's designed extensible.
1:19:03 So it doesn't have to be exactly your dialect to their dialect to SQL. You can plug and play or put in different pieces of what you need for the query plan. the data file format stuff we've talked about before, right? Instead of just having your proprietary format for your data system, you want to be able to have a a binary for file format that a bunch of different systems can read so that it doesn't matter whether my data was generated from from this library or from that library or it's in this format or that format, my my query engine can still process it no matter what.
1:19:32 And I think we again we covered a lot of these already. Parquet and or probably the most famous ones. Uh and then we have a paper that came out this year uh with or last year with one of my former students who's at Shinua and then the guy that invented um uh Python pandas and Apache Arrow. We basically showed how orc and parquet are like from 10 years ago and for modern hardware doesn't make any sense and we're working on a new file format.
1:19:55 Execution engines again we'll cover this more in this class. Veloc and data fusion are the main ones. Intel had their own thing that didn't it didn't pan out. All right. So the the main take away from this lecture is that the the again at a high level analytical data processing on the cloud or a distributed system is basically the same as as a single node. It's just again I have this intermediate step where I got to care about where the data is, what form is it in and where does it need to go next.
1:20:22 And then that shuffle phase it seems like it'd be introducing this extra step or to move data around. It's going to be expensive but actually makes your life a lot easier. Makes things more flexible. And I I can show examples of how uh data bricks and uh uh bigquery use shuffle digital optimizations next class. So there's a lot of money in the space. Uh there's probably more more people need transactional data systems than analytical data systems because think about it like if I'm a new startup, I don't have any data.
1:20:50 So I I got to get data into it like people using my product that you you need an operational database first. But the the the people that already have existing transactional data systems, they're very risk adverse because if you break the transactional data system, your company can't process any orders or do anything. You lose money. And so it's very hard to dislodge like the oracles of the world in transactions. But in terms of analytics, this this space is very active because companies are willing to try anything now.
1:21:21 Because again if if if my if I'm running terod data which is old system is very expensive and I try snowflake out or some other system new analytical system out and it doesn't work work out I still have my terod data running and I can still just use that. So people are more willing to try uh analytical systems in in environments or in larger scales than transactional systems and they're willing to pay a lot of money for it but it just makes everything much much harder. Okay. All right.
1:21:50 So next class we'll do the final review. Please come with any questions that you may have. And then I'm going to cram try to cram a single semester into one one one hour lecture. I don't know if it's going to work. I don't know whether you're going to remember anything. Uh but we can give it a try. Okay. All right guys, see you. I need something refreshing when I can finish manifesting. Too cold. A whole bowl like Smith and Wesson.
1:22:10 One court and my thoughts hip hop related. Write a rhyme in my bed intoxicated. Lyrics are quicker with a simple mocker since I'm a city slicker. Way waves are pick up rhymes I create rotate at a way too quick to duplicate feel a breeze has escape mics at Fahrenheit when I hold it real tight when I'm in flight then we ignite blood starts to boil I heat up the party for you let my girl run me my mic down with oil record still turns with third degree burn from one man I heat up your brain give it a suntan so don't just cool let the temperature rise
1:22:46 to cool it off with [Music]