H2O internals

Scale By The Bay 2018: Cliff Click, H2O internals

Recording: Scale By The Bay 2018: Cliff Click, H2O internals

hto is open source it's all open source yeah yeah and I don't work for them although I wrote the platform but I left a company like a while ago now like three years ago so I've been doing low-level performance work for like 40 years HPC computing and device drivers and all kinds of they're the core guts so the hotspot JVM compiler stuff and language implementations I have a lot of really in-depth low-level technology you know work and a lot of stuff I do is very focused on performance so h2o it's machine learning at scale it remains one of the fastest tools for doing ML it's been when I left it was maybe 10x faster than spark for doing the same kinds of operations and held a lot easier to set up the cluster and somewhat more difficult to program as a developer which was I think mistake a straw dead and I'll claim at least some responsibility for doing that because I found an easy to code to but not everyone else did best to breed algorithms these are bleeding edge state-of-the-art straight out of the Stanford math guys like whatever was the latest and greatest we did it with all the bells and whistles strong focus for data scientists enterprise work integrated with Scala Python are you could go sit on h2o cluster running Scala instead of running spark for instance it's cluster computing and this talk is about clustered computing so what can you do with h2o anything that you map from an A to B big inputs big outputs it's not just reductions filters translations conversions as all kinds is there's 100 primitives plus everything you can do with combinations of primitives and all the kinds of reductions you might want to do including you know medians and means and rolling cumulative whatever kind of things group I sort merge drawing sort of the whole gamut of what you do when you engage science is all sort of available and it all runs sort of its same kind of speeds it's all is typically memory bandwidth bound so this is intended to be because alexei right there said hey do this and i was like that's a sky conference and he said no they love it okay this is a low-level talk this is a hacker systems engineering talk this is not about Scala and functional programming this is not memory bandwidth right loads and stores and happens before in parallel and distributed in fork/join all right this is about how to get performance at a you know memory memory bandwidth speeds for everything you do Fortran like speeds everything you do so anything you want to do on a big 2d table you can read and write it like it's a big 2d table and anything nearby and you're guaranteed that it's basically going to run at memory bandwidth speed so you take your data volume divide it by your memory bandwidth of your cluster and that's basically the speed you get out you're almost always bound by memory bandwidth not by anything else the data gets compressed that helps you get you more data per memory bandwidth your data is limited to things like numbers and times and strings and stuff like that sort of a set of low-level primitives that's geared for tall skinny data so less than a thousand columns basically it's instantaneous ten thousand columns works pretty well hundred thousand some things work okay and some things they're parallelization it's done the wrong direction but after the the table long ways have tested it at trillions so I've done you know seven terabytes of network intrusion data and done a logistic regression in ninety seconds right so unbelievably fast compared to what you can get elsewhere as I said most simple Java does work so I'll show some more that in a minute it's a very clean MapReduce paradigm it's not the Hadoop MapReduce it's just a sort of hypothetically clean everything that you do goats parallel distribute for you under the hood reads writes of pins reads go the same speed is playing Java array load so say reads you you you it works like an array you have a 2d array I happen to will scale you out distributing you and you will almost always be memory bandwidth limited if you're doing something with neural nets you can get CPU limited you can do writes in parallel and strewed and scale out but writes have to follow the the JMM model and gets a little slower so i'm going take a look at how I lay the data out this should this is an old talk which see once because she does but because it's kind of fun to look at how how you these things work it's somewhat similar to what sparked us and then somewhat different some key areas so the common unit of work here is a vector which is basically an array that happens to have more than two billion elements like a Java Raisa limited to so the length is a long it can be trillions of elements right you can get and set random elements and there's a missing value because that's crucial in the data science world and it's variable size you can append and the append by the way does parallel and distributed as well so it's not you can append in the middle if you like that all works so a vector then is a giant array you can conceptually think of it as containing Java doubles but it's compressed so it's often not actually containing doubles but it has the same properties the length can be way more than four billion elements and if you walk the array linearly you're gonna be guaranteed to get the full memory bandwidth speeds be like if you write Fortran or C code for performance but you're writing Java code it's actually split across the heap tall ways this way and it is in the Java heap it's not off heap but it'll be split and so all the different jaebeum's will hold a chunk of the data and we use the standard old school stock GC with good performance because GC has a time that's dependent on the count of objects and we have very few very large objects all the data is held in small count of megabyte to 10 Meg sized chunks of arrays as a consequence the default old-school GC can handle I've done it with like 500 gigabyte heaps and modest pause times like reasonably for full GCS it's all pretty reasonable but of course you don't have one vector because you doing data science you have a bunch and so there's some sort of our frame it's our data frame it's a it's got some header stuff that's cheap and tiny and can be replicated around and it's you know destructive arrays model versus an array of structs we build essentially a struct like thing from a collection of columns and you can manipulate and add coms andrey sort them and do them anything you want any water because it's just a point of manipulation at the head but if you're going to you know within here you can randomly access but there's a guarantee that the date is aligned going this way across the columns so that when you're working with one row of data it's all local to that machine and typically be local in the same caches as well all be cashflow local as well so that operations within a single row I think I even talked about that here will be you know as you expect it sort of memory speeds to caching speeds within a vector it's chunked which is broken up into 15 a thousand and a million elements kind of varies on the compression ratio and it's set in byte arrays and held in the heap in that way and that means getting data out has to call a function it's not an array access it's actually a function at and set and that does the decompression and compression and going back in and the compression might add some overhead but typically your memory bandwidth bound and you're drowning an extra x86 cycles to go do work and so the decompression is basically entirely free like I said it's aligned within it one within one machine so you can talk about a row and talk to all elements in the row like you're fiddling with a struct and you can read and write anything in here and you can read and write things that are nearby and it'll all run at the same kinds of memory bandwidth speeds so when you go do some work it's usually you're gonna grab a chunk and you're gonna walk over it and it's going to be you know the size of this chunk has to be big enough to cover the overheads to launch threads to do the the parallel computation and small enough to get fine grained parallelism out and the code you write is actually MapReduce I'll have examples coming up but it's single threaded code you don't do anything we distributed or don't think this skate with with locks or synchronization or anything like that you don't hang with data placement there's no data placement here there's there's no questions about how many resource blocks whatever it's all handled for you so I said you do you know you you walk one chunk you write code to walk a chunk but of course it's scaled out automatically behind your back and everything runs in parallel until all the work is done and htl handles the internode communication that's necessary and it handles the roll-ups and the joins and all that kind of stuff that happens an F J stands for fork/join that's the underlying technologies Douglas fork touring for doing the load balancing of the bits so there's a taxonomy here there's a data frame which is a two-dimensional frame from now it's very common in the data science community or in the in the dev community a frames within a data frame there are vex Caesar arrays vertical columns which are broken up into collections of chunks which when you code to you actually don't ever see at a clunk check chunk happens to be a collection of a thousand to a million elements for an element conceptually Java double but it's compressed but also there's a lot of interpretation layers added so you can have enumerations or factors that will behave like a factor but under the hood it'll turn into a small integer value that will then be compressed and folded in as you expect and you can get a row of this as a common unit of things that you you manipulate okay so it's a MapReduce paradigm is the main workhorse and a map is a map from Taipei to type B and your input is big so maybe trillions of rows your output can be big or be small so you have a big output and you have a small output big outputs are going to stay big in the cluster they're gonna be distributed around the cluster because a too big to fill in one node that's the whole point so common case you filter and throughout one percent your data you kept 99% it's still big reductions take two B's and make one B and this is where you get your your vector roll-ups your sums and your going to compute mean it's a summer thing and divide your standard deviations and your min Max and Rolly cumulative averages and whatever the how you're gonna do that all happens in reductions the reductions are not timed like Hadoop's reductions the maps and reductions happen interleaved and the reductions happen immediately it's on as soon as soon as two maps are done we'll see an example here in a second on that and as a consequence reductions are typically like invisibly fast there they're just disappear and the noise most your time is spent doing maps because you have to actually touch the data that's where all the time ends up going so the the basic coding style is you extend an instance of mr tasks mass-produced task which is internally called mr task and you override MapReduce or neither or both or whatever and you say you know dot data and go and it does what it's going to do and you get your results back in your original instance so here's a layout of a cluster which has some data distributed and I'm showing eight units and I built it in a logarithmic tree which eight doesn't neatly fit seven wood and on purpose just to make it a little messy and I've made some new guy and I said go and what he does is he takes the task and he he knows he doesn't have all the data so he goes to other elements so the clustering hands the task off for them and said hey do the tasks on your pieces of data and again and again and it's a logarithmic fan out across the time so the time to launch is depth of your cluster times the network hop which is generally inside of a good well-run data center in the low milliseconds in the microsecond range and in fact the actual time to launch for h2o on a nice cluster is in the microsecond range to launch a job so you can have tiny tasks and in millions of them that's all good within a node it begins to do the same kind of divide and conquer you have a task you have a lot of data you start breaking the task up to do half than half of the data in half and half and half and half until you get enough data that's big enough to worth actually starting to do something and that's a chunk and then the tasks go and run and compute the map call on a chunk of data so the coding paradigm I'm going to show you in a minute here but you're basically writing code to run over a chunk but you don't actually specify the bounds the upper and lower bounds so it looks like you're walking over a two-dimensional array and then I'll pick the bounds for you under the hood according to where the data is and what the pieces are when the tasks are done pair by pair the reductions are called incrementally on-the-fly and anytime two tasks are done the reduction happens immediately you know within the clock cycle afterwards the reduction happens and again and again until you have done all the reductions on a node and it begins to do a network thing back up the food chain and the reductions roll up until the end results back in your instance and for just a simple pass over a data that may be some numbers of gigabytes on this kind a cluster you could expect this to be done in milliseconds because they don't take him long to walk the in or out in any direction here if you were terabytes it's gonna take longer because this is the volume of data is bigger so what's the code look like so you make an instance of mr. task and you over lie override map or reduce or neither or both in this case going to do sum of squares the map says map from a double to a double squared and this reductions of sum and so I want to do three sums of squares I can have stateful objects for which I'm going to do individual sums and the reduction is going to go across so in this case I'm summing three things we're doing all the the you know first pass of simple linear regression here so it's it's not the same kind of coding style that you get out of spark and that was like I said it's sort of the the downfall from the marketing point of view the company's still in business and still selling stuff so you know you're welcome to go download it's open source go play the stuff and make it go it's usually faster once you get to this point in coding making things run it's quick and easy so the actual the actual code you're right includes the you can write this style but then I have to do a lot of work under the hood to get you to this style but you can write this directly and generally people do because there's another efficiency hack for doing the loop here but this is just here's the loop over the chunk so this is for I equals 1 of the chunk length do the decompression by fetching from the at values and then do the math so this identical math I added the the wrapper to go touch the data if I had loop invariance and people frequently do they go above the for loop and the maps instead of passing the raw data and it's passing you the chunks and then this is going to hit your your Fortran speeds every time here's a case of doing a filter or I'm just collecting a boolean and then counted it because I was counting people under age males males below 17 right whatever it's going to be here's the filter where I append people who pass my filter and the append is distributed and scale out so I can have a billion people in the in the map and the output is however many sub billions happen to pass the filter and it's also distributed around the cluster in the same place as everyone else but this is all done you know same kind of speeds as before and the end I make a new filter I called you all and I get to get my my answer back right there so here's a histogram here's an array I want to accumulate count of cars by age so somewhere in the world I say I have some counts of different kinds of cars and ages I'm looking at and I'm going to go walk over people and grab every car and get their age in the car and do a plus plus and there's a couple things to look here the plus plus is not synchronized it's a plus plus it's a Java plus plus but this code is going to run on a private chunk of the data with a private array because I said new in the map call because I said new and a map call it's it's private to this chunk I have to I don't have to do any synchronization on it it becomes single-threaded right access so it's all cheap and easy to go get but because I'm writing it once per chunk I end up with however many million chunks I have so I need a reduction on it so we got you know utilities to do pairwise Dax be kind of adds things here's a slightly different example going to count uniques I have some billion things Island talent uniques on and some in the world I say I have a hash set that's concurrently safe so I can add to it and then the reduction says I'm going to put one into the other using standard collections put all and so I just say new uniques and go compute on my visitors and count the size and that will give me you know the set of uniques and the count of them um one of the things to notice here though is that they put the new in the class constructor instead of in the map call that means it's shared across all instances on a node and that means it's usually read-only this case I'm writing it so it has to be concurrently safe because it will be written to in parallel but only on one node every node has their own so that's the reduction is necessary so the code runs distributed it runs there's no management needed for doing I owe and machine resources you don't say new threads you don't spawn anything there's no walks there's no system exit there's no concept well the concept of global and Static variables is a little different because a global static variable becomes a node local in stat all right and then you can have different kinds of input and outputs state small state is typically that's read-only styptic we put in the constructor and you have some initial conditions you set up and you just set them as variables in your constructor and everyone gets a copy of it and it's all good things that you're writing to that are global or going to be done with the reduction because that means that everyone's writing which really means that everyone's contributing and how is the contribution done that's a reduction step so you have a reduce and if you have big state it's either read or written or both and that's going to remain distributed it doesn't come back to a single point where you can read the big value the big value stays out there in the cluster somewhere and then you run a step and you run a step and you run a step and like I said the steps are really fast so you can get a lot done there so code runs distributed in parallel basic without effort you write single threaded Java code and does the right thing single threaded coding style there's no concurrency there really actually a good really excellent resource management there's essentially no knobs needed for g4g CS or CPUs or data placement there's no hot blocks there's no hot walks to launch an h2o cluster or you simply say Java DES jar and you'll get a one node but if you have ten or a hundred that comes up in seconds and there's no other layout issues going on you load a data start doing map produces and you don't have to think about how things are set up in any other way the code can get really complicated and here's like an ugly one and what I'm really showing here is that in a show as of like five years ago there were there were 250 odd classes doing different kinds of MapReduce no more appearances as a Lloyd's pass for mister tasks MapReduce and this was written by a mathematician and sort of straight up without any help from for he was a mathematician who was learning Java for the first time this and did Cox proportional hazards and a couple of other fun ones because it was easy to write the code and then it just up and runs at you know at memory bandwidth speeds and that's sort of the core guts I'll go one more round and I'm gonna stop early for QA because I'm gonna stop early to go to the panel moderation that comes up right promptly at 5:00 so we're gonna run to the next one so just like a high-level overview look at the block architecture there's these parts and how they connect and I'll break the parts out but there's a part down here that sort of connects the clusters together and talks about how clusters talk to each other and includes things like the notion of what a node is in a cluster and the serializer we have a really fun sterilizer it's easily the fastest serializer I've ever seen and I tested a number here including everyone who claims that were fast and we were as fast or faster every time and you don't have to do any sort of setup to use it it's really slick now it's only going to go node to node to node you can't put it to disk and back directly because it doesn't carry that much information there is a front-end for connection with a web server there's a lot of ways to connect including R in Python and a browser friendly interface and and reading down a disk there's some MapReduce and distributed fork/join to run your algorithms and there's a datastore which is a distributed key-value store which is another really fast piece it's kind of fun so what's the this block this is what is a node there's a proxy internally but how do you serialize there's a thing that does cogeneration to do serialization there's fork/join there is an auto buffer which is a reliable communication protocol for both tcp and UDP and/or disk and or a couple other things in particular if I talk to s3 at full speed s3 decides it's having a DDoS attack because we're too fast at getting data and manipulating it and so s3 starts to cut TCP channel connections because he thinks he's under a DDoS attack and we have to unwind that we have to seek to an reopen stall way too do a thern s style exponential back-off till s3 light calms down reopen the cluster re seek to the offset restart the unzip in the middle of getting cut off and whatever the hell you have to do there that all alter buffer does that right it's a reliable remote procedure call for TCP and UDP and I can by the way break most revs of Linux make the Linux kernel think it's undergoing a DDoS attack because I'm too fast at setting up io channels and so I had to handle that too the Weaver this is like yet another Java serializer really but I don't need for instance to know the schema in advance so if you're writing some Java code and it's going to have to go over the wire to another cluster and it has objects and things and bits and bobbles in it you don't have to do anything at all you just use it and I you don't tell me anything about it and I'll like figure oh you're touching this thing and has to go over the wire now then I have to serialize it and how do i cyril is it Aiko Chen on the first touch I come up with a little 2 byte ID that says here's the type for this guy I handshake around the cluster hey guys if you see this - bad ID this is the type of the thing and we will use whatever compression makes sense for this thing and we'll do all kinds of heavy compression on the jaw on the object and especially on large arrays which typically in the data science community are full of uninitialized and unused pieces of it there just grow and shrink according to whatever happens you find a lot of it's all leading nulls it's all trailing dolls it's all small data in a long array where the hell there's a lot of fun compression could do and we end up being as fast as memory bandwidth to read the data from wherever compress and put it in a network buffer and we go straight to the network buffers with nioh and ship it and the other guy unwinds the exact same way so it's a fun speed test you're going to go there as a programmers to using it it's completely mindless you like don't know unless you try hard you can't tell that it's going on what else there's a key value store it it has the job of memory model and exact consistency and can cache in all circumstances and a cache hitting git will happen in a few nanoseconds because that's the time of a hash table lookup and streaming puts will go at network speed because if you're streaming puts you're actually right big data as fast you can around the cluster and you're limited by the network and supports transactional memory the big data is stored in here chunked in compressed one of the fun things then is the code that runs on a node typically looks at only the data on that node but if you have edge cases cuz they're writing a stencil calculation then you walk off the edge and you need nearby data nearby data might be on another node and then the key value store kicks in brings it local and caches it locally and you know there's there's an efficiency game there that everyone sort of wins on and then by the way the key value store is another it's a key controlling component because as a programmer if you're writing in sides of h2o2 go figure out how to make some algorithm do whatever how you want to do you typically use the dkv to talk node to node to node because it's exact and consistent and fast so it becomes easy to use to to write algorithms in so it doesn't only do the data store it does all kinds of stuff so algorithms then build up from having the data available locally and doing the distributed fork/join and above that you build the MapReduce and above that comes the algorithms that built on that and you know the MapReduce is the workhorse there are other distributed computational paradigms active in h2o and sort of straightforward to write but most the stuff's done with MapReduce and then the machine learning algorithms are built mostly on top of MapReduce and then there's a front-end which supports a rest in a JSON web server and that's your API for external things using whatever you want to do with an API there's a syntax for doing expression evaluation that you can pass through a URL so you can make a URL that says go do math and then get a rest result back this is because the math was done and here's what I got first class functions can be passed through the URL to map over the big data it's tightly integrated with our and Python and flow which is our sort of web interface that's sort of a browser interface is sort of conveniently easy to like do I don't say stupid data science point-and-click data science I can click eke licky licky to load a file that happens to be a terabyte in size and I can click licky licky to inspect columns and get all roll-ups and I can say clicking a button it says build a gradient boosting machine model on this and look at the AUC curves and stuff that comes out and it's all easy and it's limited to what you can do because that's a point-and-click interface but another hand as a first pass cut of what the hell is this data and what am I looking at it's like really handy once you go past that you like to script it and that's sort of easily done with Python arm and I think that's that's I'm out of time so I'll take QA for like five minutes and then I'm gonna be late to the panel just done with the talk so the Map Reduce takes a function that it's going to be mapped across things yeah so that the syntactic help is not present but the actual way you write the code it's new MapReduce open curly you know public void map and and and then you write the map call which if you're just doing the guts of an inner loop it's just like a lambda for first-class lambdas yeah right so so the the spark guys see it more like what Python would look like as an API you got a ripple and you can go say do a bulk operator one after another we I left before we did that that was sort of a thing to go do I don't know what its status now but I don't believe it's present you can use Python directly and have it work on a terabyte cluster behind your back at the speed of the cluster on the terabyte and it kind of gives you the same kind of feel but it's a different it's a different mindset if you're comfortable coding in Java and in Scala both the jump here is pretty small yeah yeah everyone asked this question there's not time to do justice I know exactly what's doing I'm keeping the the memory footprint really really small I don't make a billion tiny objects for everything on the planet and as a consequence I don't suffer the costs of having a billion tiny objects so the class I gave yesterday the first session in the morning I showed walking through a large dataset where by default you make on an object for every little thing you bring around getting rid of that allocation gives me a 5x speed-up so it's just sort of stupidly easily like I keep primitives as primitives but keep them small if they're small like I said it's compressed yeah okay so you think floats and doubles don't compress but a lot of stuffs not actually floats in doubles it's things like age and it's sexes of male or female and don't know and stuff like that and those kind of things compress like all held are gone right so there's bit vector you know bit set compression and there's running concluding and there's combinations ease and bias and offset because the whole 20-odd compression strategies and they're pulled out chunk by chunk by chunk and the best compression fit is pulled out and as a consequence the data set shrinks and they don't do the overhead of making it a capital D double capital all long and the combination is like I have a tenfold less size and memory to hold the same data and that's also tenfold less TC time and tenfold less bandwidth when you have to go do the math and that's the 10x beat up um that would be h2o they have had a lot of work done to do deep learning on the GPU so there's a really nice distributed scale out deep learning model that they've been applying to the GPU LAN to do distributed scale out GPU deep learning and I'm not I that was started after I left so I don't know what that status is I would go talk to h2o folks you know it is what it is [Music] um I could peel it out sort of straightforwardly it's that you need much of this base layer to pole sterilizer off because there's a cluster management first it says I have a cluster and then there's an agreement amongst the members of the cluster of what how we're gonna serialize an object so the first time you say take an object and and do something with it what you don't see is somewhere into the hood somebody said a dkv dot get of the key for the object so you had to put a key on an object but that just happens that get called did that first touch look and say do I have a serialize err for this if I don't I go come up with one and I get agreement around the cluster this is how we do it and in that set up is what you want and that's what makes the sort of magic work there's two things that come out of the sterilizer that are that you can't dodge there's a first touch notion which will test whether or not it can do something or not and then there's a what the hell am i doing has to be figured out as what it means to do serialization in the first place so there's there's agreement around the cluster and there's I touched it once and I that time I touched it I can decide I've got a sterilizer I'm it code for the serialize all right I met you custom piece of Java code to go run the serialize ER on an object that does unsafe to peek and poke its guts and that's where the speed comes from in that sense ultimately he just takes the bits and the object and runs it through the compression field by field by field and pukes it back out to a network byte buffer all right so I have a lot of fun stuff I can talk about here but I'm doing a panel now so so I'm gonna go bail thanks [Applause]