Parallel and distributed big joins in H2...

data.bythebay.io: Matt Dowle, Parallel and distributed big joins in H2O

Recording: data.bythebay.io: Matt Dowle, Parallel and distributed big joins in H2O

So just a a bit of background. So as you can tell from my accent already, I'm British and I moved across to Silicon Valley last year in March. And the the background is I I was the main author of the data table package in R. And the main thing that does is it doesn't use hash tables. It scales up by using um fast radic sorting. And but a problem with that is it's uh it's single threaded and it's single node. So I saw H2O present at one of the R conferences and they presented and said that well they've already made it made a data frame distributed and massively parallel. So I thought well this seems to be an ideal environment to scale up the the data table algorithm and it's just been an experiment to see how far we can do how far we can go

So um the presentation's all uh online and uh there's very little slides. So I'm connected over VPN over the Wi-Fi here at Galvanize to the H2O office and you're looking at our 10 physical node cluster. Uh so the the the windows in the middle are 10 256 gig RAM 32 core machines. So we have 320 cores in total. Uh these these bars here are the top that are running on each of the boxes to see the network traffic going between each of the boxes. And these these columns here are small numbers are just for my benefit really. The first number is the amount of RAM I'm using on this box and the second number is the amount of RAM everybody else in the firm is using on this box. So I've asked everyone in the in the company back in Mountain View just to hold off using these servers while I'm uh using doing this demo now

So what we're going to do we're just going to do some joins in parallel and distributed. So, first of all, uh let's start up H2O. So, if we just log on to uh one of these boxes, I've got some aliases set up. Let's make this a bit bigger. Just shout if you can't uh read this. So, if I have a look at an alias of SSH one, it's just an SSH with minus X to machine number one. So the conveniently the machines are just called uh machine 1 to 10. So if we log on to that guy and have a look at the files there, there's H2O

It's just a jar file. That's it. Everything's contained in one file. You can download that from our website or uh it comes as part of the H2O R package or it's part of the Python uh H2O package and you just download it and then stick it onto each of the servers and and essentially all you have to do is uh Java minus class path of H2O.jar um tell it how much memory if you want but that's essentially it. And when you start that on this server, it'll send a broadcast message with this name that you pro provide to it. So we'll say Matt D uh 42 just to make a unique name. And as long as you give the same name to the other JVMs on the other boxes, they'll broadcast to each other and then hook up hook up with each other. You don't need to set any environment variables in advance to make the cluster um talk to each other

So rather than do that on all 10 machines and copy and paste a command like that, I've uh already got a an alias set up and it the alias takes an argument and an environment variable which is hosts. So these are 10 machines. Let's have a look at uh the start cluster command and it's just a for loop and it's for loop for loop in uh the hosts start ssh and start java passing it the java command. So you can start h2o with a yarn environment and other guey environments as well. It's just that I'm doing it manually because I'm a hacker in quotes and so I like to control the arguments and know what's going on and not have a layer between Java and yarn whatever yarn's doing. So I'm just going to typeccl and that's sending the command off to start Java on each of the on each of the boxes. Hopefully it'll come back with a message saying it started the cluster okay on the on the 10 machines. Is the text big big enough that the Can you see read that? Uh Scl

Now we've got uh you can do this from R or you can do it from Python. Um I'm just using R because I I know it better, but it's just the same in Python. And the way I do it is just with gedit and it sends F5 just sends the command to the last window. So if I click that one and then click this one and then click here and then press F5, it sends R and starts R. So I can just I like doing this because I can send Unix commands which I do quite a bit, Python or R or Scal Scala or whatever's running at the bash prompt I can just uh send with one keystroke. So R started now let's um start H2O. So there's the library loaded and let's connect to the cluster. We can connect to any one of them

There isn't a a main node like a driver node. Uh there's no name node in H2O. So we can choose any one. Let's just choose uh machine number two. And uh 55666 was set up in that alias. Uh so that's my um port number that I just chose. So we're connected to Mr. Mr

2. We've got 320 cores. We've got 10 nodes and we've got uh two gig uh in this cluster. Uh sorry, not two gig, two terabytes. So let's um load from HDFS. So let's load up uh 1 billion rows. Uh so that's just this command. H2O.imp import file

And you just pass it the HDFS file path to the uh file. So let's put this in the background so we can see it working. So, we've got the the 10 nodes uh loading the file in parallel from HDFS. And I've moved the windows around here so that uh to put the this the bash prompt at the top of the screen so you can see it from the back. So, Mr. The machine six here is the one that's really active and that's because that's the HDFS name node and it seems that data's flowing in through the name node and so we can't it seems a bit strange to me that this is how HDFS works and there's map R as well which doesn't have a name node. Um, so that's an an alternative, but uh it would be nicer if the data was stored in some fashion where H2O could know where it was locally and just load it up locally on each node. Anyway, that's finished

That's one billion rows and loaded up. Uh, what do we what we name that symbol? We named it X. So, let's have a look at X. So, comes back instantly. It just shows the the top six rows. So it's just two columns and it's random integers uh sampled between uh one and 1 billion and uh the key is this uh so that's the key and we just got some random data in X positive positives and negatives and this is about uh 20 gig something like that. So let's load up load up a second file a second file of uh 1 billion rows. So that's going to be the same format uh a same same key and random data randomly sample between one and one billion

So it's a very high cardality test. So you got lots of unique keys. Most of the keys just occur once, some twice, some some three times depending on how the uniform distribution has uh picked those numbers. Let's just put that to the background so we can see the uh so the clusters got a bit of latency there probably because of the Wi-Fi and it's finished. Okay, so let's look at why. So why is another table 1 billion rows and let's join them. So it's just uh is it how many R users are there? Okay, good. A few Python users about the same other people SQL

Uh good. Everyone knows SQL. So the the base syntax similar to Python, similar to pandas. And so we've just got H2OM merge. uh the left table, the right table and in this case uh the method radics is this new method I'm showing it's in test so you need to pass method equals radics so let's just run that that comes back instantly because it's lazy so we need to do something with it let's just print the dimension of the answer so it's so it's running now and we can see it see all the cores are busy and we can see the data is being transferred between the nodes at the same time as so we use want use when I'm when I'm performance tuning this I want to use the network and the CPUs at the same time as much as possible. Don't want to just use CPUs and then the network CPUs then the network. So it's proceeded through it's finished I think. Yep

So it took 22 seconds to join that very high cardality two tables together. And let's have a look at the the result. Just look at the top rows. though we've seen that key key zero was uh picked three times in the left table as it happens and once in the right table. So we see that the right table value is being repeated three times to match with the three values in the left. Uh key value number two only occurred once in the left and the right. So just one row comes back in the result. Key value one wasn't picked in both tables

Yeah. All right. I have a dumb question. I'm riveted by this, but you say the blue and then the on the left. What where's the network IO and where is the CPU? Oh, sorry. The middle these middle blocks blue means idle. Those are CPUs. Uh green means user time

Red means system time. So this all these middle blocks are just CPU CPU and then these these boxes over here this this this uh every time you see a white bar going across horizontally that's network traffic there are multiple network interfaces on each of those nodes so that because there are several white bars yes so these are all the pairs so here here it's showing you uh this machine number seven to the most it's sorted like it's like top but for it's if top So it's the most active pairs that machine seven is talking to. So machine seven will be talking to nine other machines. One network interface on each. Yes, I think so. Yeah, they're all in one rack with one switch at the top. Yeah. Uh and so that's that

So the the uh that joins run and let's have a look at the bottom. So we can look at the the last few rows. Now if anybody knows how to do that in Spark Scala, just give me the last three rows. Please just let me know because I've tried and tried and tried and I can't do the same as uh tail. Sorry. Uh take right sequence or there's a function take right number of take I've seen take and take one word take right what one take right okay I'll look at that thank you and then what else can we do so this is uh this is just like a an R data frame so it's ordered in memory across the 10 uh nodes. So we can look at the let's have a look at the 500th 500 millionth row. 1 2 3 1 2 3 got the wrong name

So answer one. So that should come back with one row pretty quickly. Um you got the key value is pretty much in the middle between one and one billion. And then you've got the two data values from the left and the right. So um so you can do left outer join, inner join, all the usual things you can do with uh joins. So let's scale up. That was only really um 1 billion. 1 billion you can do on a single node, right? It's it's 32-bit addressibility

we could do a one bill this 1 billion join could happen on one node. So let's um let's just go ahead and generate um 10 billion row tables. Now if I just to speed things up rather than loading from HDFS the it'll be loading 200 gig on the left 200 gig on the right um and then creating because it's three columns in the result that would be 300 gig result. So we're going to create uh 700 gigs of uh RAM in this test. Uh so it's going to take quite a while. So let's just create some random data uh using a fra a random creation function that we have in H2O. And you just give it the number of rows, the number of columns, and the the fraction of different types of columns. Uh so here we're just going to say just two columns between uh one and 10 billion

and let's just go. So, we've Let's move it down so we can see it happening. So, it's uh started 3% done. Let's just watch the system as it's creating. So, it's the way that this is just creating random data. So, it doesn't really need to be fast. So it's single threaded on each of the nodes, but each of the nodes is happening separately. And the first time I did this, I I did create a 200 gig file on HDFS and I used ORC in the same way that I'd done it for um the smaller sized files

And the join worked and it returned the data, but it returned 50 billion rows as a result, which isn't what I was expecting. I was expecting just over 10 billion. So I looked into it and looked at the algorithm and the way it was working and how the way it was supposed to work all kinds of bugs it could have been and it's quite difficult to debug a large data problem like that because every time you do a query each of those queries takes some time itself. So eventually it came down to the fact that it was actually doing the right thing and the result was correct and the input data wasn't selecting enough random keys and the randomization algorithm in ORC didn't have a great enough period. So it reached a 1 billion period and started to repeat its random number generation. So it wasn't picking all the numbers between one and 10 billion. So I changed and uh what we use in H2O already but I didn't know until this point was the um this uh PCG um random number generators. So this is what H2O uses and I just hooked it up with their C level um they've got a GitHub and you can get download the uh C code

And I just hooked that up to be able to generate these 10 billion row files with good randomness. And uh and then that worked and returned the 10 billion row result. So that's a good uh website if you need it. Okay. So where are we now? So we're almost 96% done on creating the uh 10 billion rows. So there's uh there's x. So we've got let's see let's just make sure. So we've got one two three one two three one two three

Yep. So we've got now the way the rand this uh function is creating um random keys between minus 10 billion and plus 10 billion. So let's just leave it like that for the moment. We can have negatives in the keys if we want. So that's fine. Um and let's to tidy up just to make it easier. Let's just change the column names to be key and x1. So we've done that

Does it by reference. Have a look again the uh the data. So there's our Let's just make sure that the the column names have changed to be the key in the X1, right? So let's uh go ahead and create the Oops. Is that running? hopefully. Right, that's creating the second table. So, while that second table runs, let's cross our fingers. There's uh let's go just through a few slides. So just a reminder that H2O is um completely open source developed in the open

Uh the GitHub page is there and you see us committing to it in real time. Always will be open source. And just to reiterate where this true radic sorting comes from from these two articles, we got the idea originally on how to how to sort differently and how we can sort uh floatingoint data and uh we extended this is their their articles sort radics backwards but we're going forwards and that's been included in R itself to make uh to make ours sorting much faster. in the in the recently in the last few weeks that's been released. Uh but it's single threaded single node and limited to two billion rows in data table. So that's been paralyzed in H2O. The data table join works by finding the order of the left columns, finding the order of the right columns. The or that ordering is the index

So no matter how many columns you have in your index, the index is always the same size because it's just the ordering vector. So no matter you can have 10 columns in your index, it's going to be the same size. And then because they're ordered, it makes the binary merge much easier, much much easier. And because there's no hash table at all and we've got this ordering, it it provides some extra uh features that we can't really do with hash tables as easily. And we can roll forwards backwards to the nearest observation and and we can limit staleness. So say you've got some time series, you've got irregular data, tick data, um internet of things, device data, and you want to say what was the uh the uh temperature of the brakes of a particular car at 3:00 this afternoon or the nearest observation to three o'clock because there was no observation at three o'clock. It just goes straight. it'll a SQL join would join to an NA because there's a missing observation

But because you've got an ordered index, then the prevailing observation before that missing value is easy to return. And we can say, well, give us the prevailing observation provided it was recorded within an hour of 3:00. If it was six weeks ago, then I don't want to have stale information come back. And that's built into the join itself. you don't need to uh expand your data with NAS and then fill your NNA afterwards because that's creating more data in RAM which you then need to fill forward and it's churning through RAM. So this approach builds it into the algorithm. Uh cardality of course it's not just the number of rows that's in the data but what's in those rows. So I'd say uh top stock tickers they're low cardality 500 is a small number millions of people medium cardality and billions of devices high cardality

So one of the use cases is you've got several terabytes of new data a day you need to ingest that join it to another couple of terabytes and decide what to keep before the next day starts. Uh that's the orc script I described which um uh which was fine up to 1 billion rows but uh doesn't work for 10 billion. And then just to show the data so as we're increasing the number of rows from 1 million up we're we're also increasing the width of the key which is really important uh for a decent test. So um 100 million rows, one billion rows, the keys are getting wider. 10 billion rows, 200 gig. And then the H2O commands are really simple. You just library H2O or import H2O in Python. Uh you init command to load the data and then merge just like u R or Python and then usual commands

So, it's so simple. It fits on on one slide. And then the scaling. Uh, so I've showed I've just uh demoed the the 30 seconds. Actually came back as 20 seconds, didn't it? In that test. That was 1 billion rows. And, uh, we're just about to do the 10 billion rows. Um, it should take about 10 minutes if it works

And then the the I've demoed this in the past, but only up to one billion rows. And so there's a YouTube video of the uh join working on 1 billion uh which is on that link. So let's uh hopefully the data is loaded. Yep. Sorry, not loaded but been generated. Let's just check. Yeah, it's got the default names. Let's change the names

Okay. So, we've got our X and Y. We've got 10 billion rows in each one. Each is about 200 gig. And we're going to create a 300 gig result. So, that's started. Now this Oh, 10 minutes left. Perfect timing

So, let's log on to that um Mr. 2 machine. And there's a log file that's being generated. So, let's have a look at that as it's running. So, it's uh Let's make it a bit bigger. So, you can see the CPUs are all all happy and uh busy. Nothing happening happening on the network at the moment. Uh so, it started

So first thing it does is it creates the left index. So basically it's just sorting the table. So at f first thing it does is it it does a histogramming of the most significant bite in parallel and distributed across the whole uh table. So then it knows how much of each bucket there is and then it can work out the offsets and it can send that those subsets of data to the right node. So essentially if you've got like think of it like a dictionary where you've got uh surnames or words from A to Zed, it's putting all the all the words that begin with A on the first node, all the letters with B on the second node and so on so forth. And then once that data is there with a row pointer to where the original data is, it then sorts locally on each uh on each uh bin which is quite different to other algorithms because other algorithms and other um frameworks they sort locally first and then do a merge afterwards and that merge afterwards step is serial right this is this is a different approach as far as I know. So it's um so it's already done that all that splitting. So it's finished now and it's finished the left index in uh 116 seconds

So just under two minutes and that's really high cardality 10 billion rows. And now it's doing the right index. So it's doing the same thing. It's split up uh it's counted the bins by the most significant bite. Uh now it's sending them to the right nodes. It's done that. And now it's on each node it's it's sorting all those uh sub pieces. This is my day

Welcome to my uh development day of trying something new and then testing it and seeing if it works. I'm an outsider from R. So I'll admit that right. Thank you for H2O but I don't know much about it. When I tried a few tricks in R to explore, I found that the memory expansion per variable was very large. That was my untrained observation was that it was so when I loaded, you know, a billion rows of CSV with different kinds of values to R, the amount of RAM that was taken seemed crazy. Which which function did you use? I don't remember. You are we in a day when RAM matters so little because you can put this together? No, no, no

So, so first of all, could you sum up the question for Oh, for the recording. Yeah. So, the question was um an expansion on the the questioner doesn't hasn't admits you don't know R that that well and your initial first um observations were that when you loaded data are just consumed RAM very rapidly. Yes, it was a CSV file, wasn't it? So, you probably use read.csv, which is the base R function to read. CS, which is appalling and everyone knows that it's appalling. It's been appalling for a long long long time. So, you just cannot use it and there's no way even if you pass all the arguments you're supposed to do, like if you tell it the number of rows in advance, if you know that from WC minus L or something from the command line, then it can get faster. And there's some more tricks that can make it about three times faster

But even that's still appalling because you're absolutely right. It does eat up RAM. So the fastest way is fed in the data table package and that reads it much much faster. So if just use that and I'm biased because I wrote that. So that's that's why one of the reasons data table is is popular because uh about three or four years ago F read was added. What's the relationship from H2O to the data table? Very little. Very little. So, uh, H2O is implemented in Java

So, this is the JVM that you're seeing which is being distributed and um, paralyzed. So, we're using all that great stuff with the the garbage collector in Java, all those parallel threaded libraries and tools. And we're using R as a front end. So when you type R and it's a H2O frame, the computation gets sent to the back end much like a database and the same for Python. So we're using the same so all this is happening in Java land. Thank you. Yeah. So speaking of Java so so just just while we're on the same topic and then so H2O has H2O.imp import file which is much which is at least 10 times faster than FRED that I wrote in data table because data tables F read is single threaded on a single machine whereas H2O.imp import file takes a CSV file like from HDFS these ones that we're reading now or from local disk and it'll read them in parallel on a single machine using all the cores and distributed across 320 cores

So that's why companies like PayPal are using H2O because they they tested F read versus H2O import file and then this is just much faster and it's why I'm at H2O because I couldn't beat it. So yeah, speaking of Java, at some point I thought that Cliff was working on a hacked version of the JVM. So what happened to that? Not a hacked version of the JVM but optimized. Yeah. So a lot of the things Cliff has so this frame here which is distributed and paralyzed is all Cliff's work mainly. So he designed the the H2O frame all the compression. So when you when H2O import file looks at the file it's already got the data rowby row right and it's very simple. It just maintains the minimum and maximum of the column that it's seen so far and that's free to compute and maintain

So things like those optimizations he's put in. Yeah. Understood. Let me try to ask it a different way. The JVM that you're using right now is the stock JVM. No, it's the Oracle Oracle stock, but it's a stock JVM. Stock JVM. Yeah

So you guys didn't have a specialized version of the JVM. No. Okay. No, you can see it. Yeah. So all the optimizations are just at the level of Java. That wasn't my understanding. Now well Cliff Cliff may have put some optimizations into Java itself because he can I believe he would have done that years ago

That was that was his positive presentation you know that he actually went all the way into the like garbage collection implementation of open JDK and stuff like that. So to me it meant that it would require a separate you know copy of Java. No I think that's he was talking about what he's done in the past in the Java hotspot compiler which is already in Oracle. Okay. And another question, what happens if the node goes down? Good question. And what I like about it, oh, yeah, the question is um what happens if a node goes down? Well, in my experience, this is a physical node in a rack, which I think companies are quite often likely to have and nodes don't really go down. What happens if Java crashes at this scale? 10. That's actually my follow-up question because the node could go down not just because the hardware because there I would absolutely agree with you software parts typically yes software things and I've seen spark going down with Java restarting and so on doesn't happen with H2O okay because it's it's just written it's just it just does that doesn't happen okay so I think there's a a penalty of the resiliency in spark that we're not paying so if you run this disj join in spark you'll see it writing to disk into the work directory, right? And it writes a lot so that it can recover if a node does go down

We just aim that the nodes don't go down and we don't pay that cost. So, it's a lot faster because it's in it's truly in memory only and we like we like that. So we're not really aiming for Yahoo scale or Google scale of thousands of clust nodes where cosmic rays can take down a cluster. Apparently that's not really our space. Okay. So it's and I I totally respect your engineering choices. That's that's fine to make that choice. Again, there's nothing wrong with it

But just sort of this obsession about, you know, hardware failures, that's not what the problem is, right? So I used to be on the original Hadoop team at Yak, you know. The reason it was engineered for extreme fall tolerance is because of software bugs. A node would go down because of the implementation itself being buggy. Yes. All the hardware going. Yes. Well, repeat. You tell me

You tell me. Well, repeat but repeating it isn't going to make it not buggy anymore. It's still going to fail. Recovery recovery helps because you know then we can recover into a state that actually doesn't trigger a bug anymore. It's in practice. It's very very hard though. Let's finish and then we can talk about it later. Um, no, that's that's a good point

But I think I think in my in my experience when I've experienced those crashes, software bugs, Spark will repeat it over and over again and it won't recover. That's what happens here anyway. Um, so that's the Has it finished? It might have finished. Yes. So that's 10 minutes and let's have a look at the result. So that's uh so remember it was uh randomly s keys sampled between minus 10 billion and one and plus 10 billion. Uh so the smallest keys are first and um and it's ordered as well. So we didn't have to do a join then order by because part of the algorithm itself joins and part of the process of that joining remember we created that index and then we used it and we created a second index and then we used it we've still got those hanging around

So if we go and do a query and filter or join along those dimensions of the key then it'll be much faster because those are knocking around which is exactly the same as what we do in data table in the in the R and the R project. So what can we do now? Well, why don't we go and select the uh it's an ordered table, remember? So, we don't have to do any kind of um SQL adding a column which contains the row numbers or anything like that. We can just go ahead and select the three billionth row. So, one, two, three, one, two, three, one, two, three. and uh type the right object name. So that should hopefully just come back. So what's the performance of that expression? See, it didn't come back immediately. That's interesting

Well, let's have a look. To find the 3 billionth row, wasn't the performance there? Yeah, that was a bit um that may be the Wi-Fi. Let's do it again. It's warm, right? So, I could have a look at that. That could be I'm just curious. I'm too. Yeah, it would be nice to have that coming back instantly. And let's see

Yep, that's all all I had. I have a question. Y questions. what you showed is good and it works and I'm sure that it works but what would you use for you? Right. So um well to be honest part of it is the the the sorry oh to repeat the question the question is uh it's great it works. It's amazing but uh what would you use it for? So part of it is that this is the most difficult test we can think of to test the H2O infrastructure. Um so we're got a high cardality test and so on. So you saw that the index was being created quickly

It was only only took a minute or two to create the left index and the right index. So you can see immediately how that can be useful is that uh you know faster building an index in a database is standard. You want to group faster, select faster. often it's long sever common uh the key is often used and reused many many times as you don't want to do a vector scan over and over again and this so this is just a way to test that it's doing the correct index and that we're getting the correct result but also if you so if you really want to join two large tables like this one use case is like I said where you have several terabytes of new data each day and you want to join them together quickly and traditional database cases are taking hours or possibly days and you have to scale out and uh scale you have to all kinds of tricks scale out much more than we're using here many more than 10 machines uh to get a get to get them to compute the result within one night before the next day starts. Um anything with internet of things with devices very high cardality. Uh people are using it for unique product codes because they are integers. Of course we can do non- integers as well strings and enumerations but uh quite a lot of data does actually have these very large integer numbers which need to be joined together. uh devices in cars uh like progressive all the GPS data that and all the the car telemetry that's coming off these devices are pouring in every day that has to be ingested sorted features calculated each night and stored before the next day starts

Thank you very much. Great. Thank you.