Scalable Scientific Computing using Dask

Speaker 1 [00:03]

Hi, I'm not giving a talk, I'm giving a workshop today, so that's why we also have desks in here. So for the people at the back, I'm sorry we didn't have more desks, but at least we could squeeze everyone in. But also if you want to work with me on the workshop, it's kind of helpful to open your laptop. I will share a link later where you can also just open up a Jupyter lab and work with desks. Short introduction to me, why I'm giving this workshop, or why this is actually in the context of my work. I'm a data scientist at Blue Yonder. We're a car-through-based company doing software service data science for supermarkets and retail industry. I also work on the Apache Arrow and Parquet project. But actually, my daily life is working with heavy data in Pandas and data that is much, much bigger as your laptop's RAM. So actually, it doesn't even fit on a single machine in the cloud. So I need to have a bit of distribution and work with Pandas algorithms, but just to be among many hosts. An important thing for people that are on Twitter, there's my Twitter handle, and it might be useful to open my latest tweet if you don't want to type long, because my Twitter handle is shorter than a tweet. Or shorter than a link that I will post. But also, yeah, even besides the link, you can follow me on Twitter. I'm going to talk a bit today about Dask, and also the library distributed behind Dask. I'm also going to explain later to you what the difference is between Distributor and Dask, because this is a really particular separation between those projects, but this also gives you a hint where these two are focused. So Dask as a library, it's there for you to execute graphs and build up these computational graphs, which is actually just a data structure telling you how to compute something. On top of these task graph building on Ender, there's also some high-level APIs like Dask DataFrame or Dask Array, which provide you with an interface that you can use Dask just as if you're using Pandas, but in the end you don't get a result. You get a task graph you can pass down to a scheduler, which then executes it. Dask already comes with two executors. There's the local in-process executor, and there's a multi-threaded or multi-processing executor to give you parallelization on a host. If you want to go beyond a single host, we will have a look at Distributor later on. The nice thing is, just by using Dask, you can write algorithms, and if you write them the correct way, you can scale them on your laptop. And you can work with Dask on a laptop, just like it would be a distributed system, and it also can run already on your laptop, Python algorithms in parallel. but the nice thing is if you have this on your laptop you can go up to one or a hundred or thousand nodes distributed in your cluster still right with pandas or numpy like api but in a distributed fashion yeah as already said there is more than a single cpu that's the thing where pandas is really good if you have one cpu have lots of ram on one machine you can use pandas you can do really quick data analysis but once your data set is bigger than just a machine your machine probably your laptop so data set bigger than like maybe five or ten gigabytes you're out of luck you need something else because it doesn't fit in ram anymore but even if you're on a laptop it's kind of nice to have parallelization because a single laptop like mine has two or four cores so um with a single thread in python and a global interpreter lock you normally can only work on one with pandas just on one of your CPU cores. As already explained, Dask is split up in two parts. You can build up and execute computational graphs. And there's a high-level API. There's DaskArray for number arrays. DaskBack is for distributed collections, like distributed lists or distributed dictionaries. And there's DaskDataFrame, giving you a distributed version of Pandas. one main thing or the one last thing i give you were slides instead of just really doing something is um people normally compare dusk and spark um there is not one that is better than the other but they do have a different focus that's the thing you need to keep in mind if you use them dusk is really lightweight and it's written in python and it really works really good together with the Python ecosystem. And also with these C, C++ things, which you often import in Python with a thin wrapper, you also can use that to parallelize them with Dask. In comparison, Spark is a JVM program. It's written in Scala. It's really good in that ecosystem. It has also its own ecosystem. And in comparison to Dask, Spark is not a lightweight. So there's things like whole process optimization or whole task graph optimization. That's a really big thing in Spark. It's really something that gives you a lot of performance there. But on the other side, it just makes some programs unpredictable. And if you really know how your algorithm should be structured, that's kind of countermeasure. But if you don't know how something should be structured and you would like to have this compiler that does some magic for you, you use Spark. That's the thing I mentioned here with high-level optimizations. But also, there is PySpark, where you can also use Python in Spark, but it's a rather thin layer on top of the JVM things. And if you use Python UDFs, they're normally typically slow in Spark. Nowadays, there is error support in the latest Spark release 2.3 with the addPenAsUDF decorator that already gives you something like 2x or 10x speedup on some operations, but there's still a lot of overhead because you're writing code in Python, which is running in a separate process from the Spark workers, which are running in JVM. Yeah. So this is the tutorial. So I'm going to switch to JupyterLab. Also, you all can switch to JupyterLab. My slide on Friday was looking here with a conda install, but given the network and given the setup Matthew Rocklin provided two weeks ago, I'm not going to go for a conda install, but Matthew Rocklin has provided similar slides to those I had on JupyterLab but they come in a binder so basically you click on that link there are the sources for the Jupyter notebooks but there's also a really big button, launch binder click that and you have to wait 1-5 minutes and then there will be a JupyterLab started with all the things you need for the tutorial and please do not yet execute anything because we have people anticipated, the cluster resources are not that big. So just start a binder, and then we're going to go through a bit by bit. I'm going to leave this now up so you can type it. So when you have started, there will be a long loading screen, which can take some while, but I'm already here in the setup. You cannot go on to the left side. This is probably shown for you like this, and we're going to have a three-part workshop here. My intention is here we go through each of those notebooks. I'm going to give you an introduction how the basic things work and at the end of each notebook there is a kind of self-exercise things where you can actually try out and ask questions and see if you can understand what Dask is doing and what's behind that. So three steps. We're first going to look at how to distribute operations on a Pandas data frame. The second one is giving you a bit more insight on how you can do low-level scheduling or low-level distributed scheduling using Dask. And the third one actually uses machine learning and how you can use Dask to use a scikit-learn model and train it on multiple nodes in parallel so you get better performance. The special thing about this JupyterLab setup here is, we have a DAS dashboard extension installed. I will explain these two later when they pop up, but when you want to get insight into your DAS cluster, you can have a look here at the different things and click on them and it will pop up. So can you give me a hands up who already has a binder started and it's working for everyone? Amazing. I haven't expected that. The first thing we're going to look at is distributed data frames. When you have pandas, you work on one machine. You can use just every function and it will work. DAS data frames are built on top of pandas, but as we also having distributed algorithms or distributed execution, not everything will work. That's the main thing you have to take care of. Even though the API is there, and it gives you the same feeling as Pandas, some things can be slower if you're not on one machine, and some things don't even work. That's basically because if you're on one machine, all execution happens on there. If you go to multiple machines, you have network communication. And network communication is that which you always want to prohibit, and which makes everything slow. But if you're going with large data, this is something which happens from time to time and has to happen if you're not really doing really dumb parallelization the main difference also to pandas is that we have task graphs so at the end after calling these apis on das data frame we will get an instance of a das task graph which we kind of visualize but which you also can use on dot compute or dot persist on to actually trigger the computation and get a result but why until you call compute or persist you're solely building up a task graph and there's nothing happening so first thing like all notebooks we're ignoring some warnings um just so they don't get in our way but in this tutorial we're going to use the new york taxi data set i don't know if it's Probably one of the most known data sets out there. But the nice thing is, it's a lot of taxi data. It's a huge amount of data, and it's publicly available. So a lot of people use it just to do demos with that. I think one year of taxi data is a billion rows or more. So it's already kind of huge. On disk, I think it's about 700 megabytes per month. So it's 10 gig per year. But also in memory, if you do computations on that, it will get a lot more. For this demo, we have the data stored on Google Cloud Storage. This all is running on a cluster of the Pangaeo Research Project, which is funded from the National Science Foundation in the US, which is also giving funding to Dask. So this is how we got a cluster so everyone can use it. But also, this is an Anaconda public Google storage, so that's also a thing you can use outside of this tutorial, but actually just for the demos here. And you can see we have one year of data in there. And the simple thing, what we could do first, is read it with Pandas just to get an insight. As I said, it's a large amount of data, so when we read it with Pandas, we're going to use the nRows parameter to just limit the memory we use at the moment. And we have two date columns, which we already want to pass as the date time. And this then in our notebook loads the data from Google Cloud Storage into memory and reads the CSV, and we get a data frame output. But this is all happening locally, not without any distributions. And because it's Pandas, execution is direct or eager, so you directly see the result here. This is trip data from New York taxis, which tells you where the trip has started, where the trip had ended, and how much the trip has cost, and some additional amounts about how the cost was made up, and how many passengers were inside. So it's actually a bit of data you can already step through, make some analysis, but as it's already huge and bigger than your laptop, it's not a thing you're going to do locally in all data. Where can we see 100,000 rows, 168,000 passengers, or simple thing, you can look at the mean trip distance. This is US, so it's 2.9 miles. And also, same thing, how many passenger miles were done. basic pandas. I hope everyone has seen this. This is just code for demonstration. We're going to pick this code up later and see how can we pass it on all data. And now this is the tricky part where everyone here please change the line. We're now going to start a DOS cluster on this Kubernetes cluster, which is powering all this. Because were more participants than anticipated. I would like you that if you could all put in a 5 here, not a 20. That gives all other people more resources, and we can all still have distributed computing. While this also tells you a Dask Kubernetes cluster, what we're actually starting in the background is distributed. Distributed is a Python artifact, which is also part of a Dask project, but it's a different scope. Dask is for building up task graphs and having high-level APIs on that task graph. Distributed is actually just a scheduler. It takes these task graphs and executes them, and it does this in a distributed fashion, just like a name called. You can do it on a local machine, then it will start many Python processes, but you also can use these Python processes on another machine and connect them to the main scheduler, and then you can use multiple machines to execute your task graph on. that's the main separation yeah you can ask questions when you want to bring it down there's this nice widget here where you can just type in 5 and it already scales down so this artifact does Kubernetes is another Python artifact which is used that it connects to a Kubernetes cluster and it sets up a DAS cluster but it also can control this task cluster. So doing manual scaling here to five will size down your task cluster. You also can say adaptive scaling, where you can say an upper and lower limit in resources, but it will also then match how many resources you actually need. Also, you can click on the link down here, which takes you to an overview of the task cluster, shows you which workers you have, but also gives you a kind of nice system overview. How many RAM is currently used by workers, and what's happening. It's really nice, but as we're in the JupyterLab, we can use the JupyterLab extensions on the right-hand side. And we have on the upper hand here is the task screen. So when you have a computation graph in Dask, Every node in this graph is a task which should be executed. And that is shown up here. And also, if you hover over that, you will see later on when we execute the first code what actually is happening there. And down here, you see it distributed by worker and when it's happening. And down here, if you submit a task graph, there are multiple nodes that are doing the same. They aggregate it here, and you get a progress by how far they already executed. So I already told you the difference between Dask and Distributed, and this here is another line where you actually connect Dask and Distributed together. You have a client to the Distributed cluster, you instantiate that, and you connect to the cluster you have just started up here. Per default, it's Dask using just the latest client you instantiated in your notebook or your Python process. So if you are using multiple clients, the best thing is to use context managers when you want to switch between clients. But as we normally always stay in just one cluster and work with the other one cluster, we're just keeping here the global variable. And that's now where we're going to use our first distributed computation. We had a look up earlier where we loaded just one CSV and only did the first part of the CSV into RAM. But we actually now want to have the whole data set into memory. And we have that on Google Cloud Storage. We can pass Google Cloud Storage URLs to Dask. But also, here's another thing you have to change. We're going to change the globbing pattern here to just match all months that start with a 1. So October, November, December. Because you only have five workers, so it's a lot more weight if you're just working with a smaller data set. And then the first line here creates a task graph, and the second line calls persist. Persist is one of those two calls I mentioned at the beginning which starts execution of the task graph. Compute is the one that already computes something and returns the result to where you executed it. when we would call compute in this scope here, our Jupyter lab would crash because it would just gather all the data frames from all workers, put it in one Jupyter lab, and would be more than our memory. Persist starts the computation, but persists the results on the workers that have computed that. So we have the result up until here in memory, and we can continue working with that instantly, but it's not on my local machine, but it's distributed along the cluster. Now what you're seeing here in this block, When you execute that, which you all can do now, you will see something like this task stream over here, and there will be three tasks down here. It will be read CSV, read block, and pandas read text. These are the three ones that will pop up here, and you will get a similar work stream to this one above here. then maybe we need more ah yeah you need to be put at 200 2015 minus one or trip data even here you can see the file names in the the uppermost google cloud file system block Okay, when you're getting zero worker, zero memory already exceeded the cluster quota. And the thing what you can do instead is you can use your notebook and make a local cluster. So replace the Kubernetes line in your notebook with this local cluster, then you should get more calls. Or you get the local calls, but you also are in your notebook, so you have to be careful with resources. So the question is, should the data appear automatically if some workers are running? Yes, it should appear automatically if you have run this line. There should be no data, but it should just return. What you can do to see if you already have data in your cluster is execute this line down here, where it says passenger-count.sum, and then it schedules the task graph, which is sent to all nodes that says, okay, sum up the data frame in column passenger-sum, But as we're distributed, each worker will do the summation on this column on the worker. And then the summation of that will happen in a reduction-wise that two workers communicate with one another and sum up. But the nice thing about this one here, it's the same line as in Pandas, but it happens distributed across your cluster. And as data already was in RAM, it runs really fast. So, 50 milliseconds just asking 20 workers and getting a result. Maybe I did the wrong thing. I inserted another data frame class at the top. So yeah, presentation fault at the top. I made a local data frame, also called it df, made a data frame here and called it df. Now, if you're running df.passengerCount, there's actually missing the compute at the end. So there should be no result. It should be only a task graph. But when you're lucky and have 20 calls like I have, You can see how down on the left side where the brokenness is happening. Now when you do it correctly, you get a distributed scalar with a computation behind of that. And now, if you want to have the real result, you can extend it with compute, and then it runs through two small tasks down here, and you get really a big result, but even still in a fast time. And hopefully, this should be working now for everyone. If you're in a local cluster, it may be even helpful for you to just put one single file name in here, and not more than one file. Or you can also use the end rows here. That will make things faster, because it will only pass part of the files. But it will still give you parallelization, because read CSV automatically splits up a CSV file into chunks. It does some guessing, and it has to go a bit forward and backward. But it can still write read CSV files in parallel, which also gives you some performance benefits. You need to persist here so that, OK. Press the wrong button. Thank you. stay connected No, it's not working anymore. And sadly now our visualization is gone, but we still can continue. We have called here persist. Persist means that we, at this point in time, already load the CSV data into memory. We're working here already with data that is in memory, and then can already compute the sum. If you wouldn't have called persist here, this compute call down here would have read all CSVs files in memory, computed the sum on them, and already thrown them out of memory again. Persist keeps them in RAM, as long as your Python variable lives here. So if you want to un-persist something, just call del on that variable, and hope that there's no reference anywhere else. Or you can call unpersist. So do you believe that data is to the memory and best to persist? The best thing is when you call persist, normally as late as possible. But when you know you want to use data multiple times, then call persist. And Dask is always also keeping track of how much data you have in memory. And based on that, it will also, if you have enabled it, persist data from memory to disk. And I'm back again when you're doing computations on that. But it also depends on your disk speed if that is feasible for you or not. But if you only want to compute the sum and nothing else on the data frame, you shouldn't call persist. And because this is a workshop, there is now some sections down here called Investigate and Questions. The best thing for you is to look at them, try to do them here, and ask questions if they're working. And also, just a big shout out to Matthew Rocklin, who is not here, but who's already given me the tutorial live support. There may be more nodes spinning up now, so the people who have a local cluster may try in five minutes to just connect to a Kubernetes cluster. And you should get nodes also that you can use a bit more data, which is a bit more fun. So are there other questions, other people I can help? These are the most basic things, so I'm hoping they work, then we can go to more advanced things later. Yeah, because when you have called persist the results are already in memory. So when you call compute again, it doesn't really have to do anything. Is task then saving the graph that associated with that call, that call chain? How does it know? TASCAL TROYER- When you look at the task graph here, this is what happens when you think this is read CSV, the task graph you you have three tasks the first one is um reading the real data from google cloud storage the second one is actually putting it into pandas and the third one is putting it into memory and you can see up in the top um you have released on memory that the two states a graph can have, or it can vanish from memory, then it's no longer gone. But it remembers the task graph, but it also keeps part of it in memory. So if you call persist, it definitely keeps something in memory. If you just have a big compute graph, it will keep some things in memory, which it thinks it needs later, and other things which are then blue will not be in memory. And sometimes things are getting computed twice, but that's okay because computing things twice is sometimes cheaper than passing it over a network. But it's also a good thing if you look at this DAST dashboard, which is the link normally if you start at a cluster or when you just call a cluster in your notebook, you actually see what's happening behind the system. Yeah? So I assume the Sys and DAST is similar to Cache and SPAR. how does it distribute the data among the nodes? Yeah, it's kind of similar. I think there's some slightly different semantics with cache in Spark. The distribution is mainly workers that have nothing to do at the moment will pick up work from this main scheduler. And if it's a linear task, it will stay also on that one worker. And the worker that computed the result will also keep the result in memory. There's one thing in Dask which is called work stealing. If some workers have more work than other workers, workers can steal work from another worker. So that even if your cluster scales up or down, and workers that are doing nothing, they can get work from other workers. Meaning that if you have one worker which has a lot of data and memory, and then you want to continue, and other workers don't have anything in memory, you will copy that data over a network, and it will continue to work from some of the memory on another worker. So, now going through one of these exercises, is everyone done or has anyone any questions? At least some feedback. It works. Okay. It works. Then we can go through this here. You can just take the pandas code. Our data frame is also called df. So we can... I think I've got the wrong data frame again. So basic thing I have to learn from now, don't call anything df. I take the data frame now, I filter on passenger count 1, typical Pandas operations, and then I get my intermediate data frame, which is actually a task graph, but the task graph has some annotations on that in Dask, if it's a data frame, because it actually knows the columns, and it knows the data types of the columns, and there's something called end partitions, that's how many partitions there are of a data frame or chunks in memory, meaning that we have 20 workers, but we have 365 splits or chunks of a data frame memory on those. We can then work, and we can do parallelization on each of them so we can do 365 threads in parallel or cores in parallel, meaning that we can even scale our cluster more up and it will get faster. Yeah, then we tell it we want to compute the mean, as this is now really a DAS data frame and not my wrongly panned data frame again, we just get a scalar computation graph. One thing we normally can call on that is visualize, because if you have a DAS, oh yeah, Maybe a bit too big. That's also a thing you can normally call visualize on that. But it will visualize your whole task graph. And if you have 1,060 tasks and you do something on top of that, that will be a big graph. So that was a wrong idea. I will pick that up later. But if you have a small graph, mainly in development, you can actually visualize that and look at it. But here we're just calling compute. And we're calculating the mean on that then. And also we can switch over to the task graph here again, where we're seeing where the parallelization happens. And here we also see that we have more than 20 chunks, because each worker takes a column of the data frame and then computes the mean, or a sum in this case, and then takes the next column chunk. And does this at least twice, or sometimes even four or six times. and at the end we have these big red blocks red blocks are the bad ones which you don't want to have but you actually have to have because red blocks say that the cluster is communicating some things over network that's the slow things but if you make a distributed computation but want a single result you need to do this so a good algorithm normally has a few amount of those blocks has some of them but a lot of non-red blocks here we aggregate all the sums together and because it's a mean probably the last block here will just divide by the whole by the number of elements in our data frame and you also can see the level of parallelization here everywhere that's white there was nothing to parallelize so the red blocks waited until the last mean block here had finished and then to communicate over the cluster so the better your parallelization works the less white space you have. Yeah? For example, my cluster starts with two nodes and I write the data set in two nodes and the data scales up to five, then I would have a lot of communication costs. Yeah, that will be because it redistributes the data which is in memory. But does it store it on the other workers once and then everyone takes it, or does it always for each operation? No, it stores the data only once and normally only on that worker that uses it as latest. So that's the thing. If one of your workers crashes, the task graph has to recompute it until that point. But as it also stores the task graph in the scheduler, it knows what it has to do to get adapted to that point. So that's also a thing a task also assumes, that if you have a task graph, it's no problem to execute it twice. You can mark things as impure, then it won't be executed twice. But normally all your code should be written that it's no problem to execute it one or three or four times and you can also see here over here by stored you see each worker and you can look at how much each worker has thought so probably in your case when you have three more and workers there should be two big blue ones and the other one should be quite small or after the first computation they may change here So, that's also the point where I'm going to mention again, compute is good because you get a result, but only do it when you really have a small result. If you have a big data frame and you want to see the whole data frame, don't pull it into your local machine, because your local machine is normally a small laptop, and it's not a cluster with one terabyte of RAM. If you want to peek into it, call head. That's the same thing you would do in a Pandas data frame. it will get you just the 10 uppermost lines. If your data frame is not sorted, it will be just 10 lines from somewhere. A different example, which I'm going to skip so that we get to the next point in the workshop, is you can call this without persist, but on every call to sum or mean, we would read the CSVs again into memory and compute everything from there on. And reading the CSVs is the most costly thing we had until now, so that's a thing you don't want to repeat. So the call to persist at this point was quite helpful in doing analysis. Also, when you call persist, this is the thing I already showed here, you can see what is happening over here, each graph on the cluster. But the also nice thing is, when you actually develop, you want to see where the bottlenecks of your code are. And Dask already runs on distributed, already with a statistical profiler, always enabled, and also aggregates all the statistics over the network. So that you get, if you go on the profile plot, you get actually a flame graph, which I think takes every 100 milliseconds or so, a stack trace, and transform it into a flame graph. And then you can see where the time on your cluster is spent. In this case here, we're spending about 20% in libSSL read. It's cloud storage using HTTPS. So that's one of the costly things. And the other costly thing is the read method in the Pandas parser. And I think we have to go down here. It's not always the best, DuckTrace. But this whole part here is a CSV reader. And so we can see that 20% of our whole computation time goes to SSL handling, and nearly the rest of it goes to reading the CSV. And a tiny bit over here, this is just a summation on the mean computations. And we even computed several means. They're still all here in the cluster size. And you can see here the peaks. The peaks are where you actually have done some work. and you can just mark them, and the profiler will go there. That's a really nice feature, because you actually get an insight directly what is happening on your cluster. And if you do use this productively, you need to be careful, because there will be a lot of profiles, but the overhead is so small that you can already look into it and see what's actually going on in your live system and what's making it slow. Looking back to the basic design of Dask, Dask is a library building task graphs. It's not a library building, actually, data frame operations. For data frame operations, it's using Pandas inside. Pandas is still a single host library, but the Dask provides task graphs that call Pandas operations on each host. I already mentioned quite earlier, just two minutes ago maybe, that we have 365 partitions of our data frame memory. this is the normally split up like here like per month or in our case per day where we have a small data frame pandas data frame memory on workers and they all work with pandas operations we do pandas operations on them but we have a task graph of task on top that orchestrates all these small pandas data frames to act like one huge data frame which also gives us just normal results when we call compute this is also when you have have an operations that's available in pandas but not available in dask there is the magic map partitions um when you call map partitions um you see that the main input value is a data frame it's actually your local data frame you have um the local pandas data frame and you can walk on that so if you know a pandas operation like maybe which you want to use mean but don't want to use dask.mean you can call dataframe.mapPartition and pass a function to it that function is then pickled, sent to the whole cluster and executed in all those small dataframes and then you return something again and work with the dataframe again you can work with it as a dask dataframe further on but you can already write these map operations so that things that are not implemented in Dask can be implemented in your code directly. We have those chunks of memory. We have these 365 chunks of memory. 365 is maybe the number of days in a year, but in our case it's also a bit arbitrary. If you have distributed data, you sometimes don't have it stored like you really want it because you do maybe an aggregation by year, by month but sometimes you want to do it by vendor which then means you will have a lot of communication especially if you're doing some kind of really complicated selects. And in Pandas the index is normally just a descriptive thing gives you some information how your data is laid out and how you have grouped by. and that's this index also used for distributing the data and when you do especially when you're doing group by operations normally group by will gather all things together that belong to a group send them over network so that they are on a single machine and they can do operations on that but if you already know ahead you will do several group by operations they're all going to depend on a pickup time then you actually should set your data frame index to the pickup time and call persist and then dask ships nearly all data over network which is not good but afterwards all the chunks you have are laid out by pickup time so that meaning that if you do an operation that groups by pickup time it will directly work on a local node And now we can see here, that's the bad thing, which we normally don't want to happen, but in this case we already know, we want to redistribute our data, we send it over network, it's costly for one time, but on all later operations we don't have to see these red dots anymore. And we can also, if you look at the printout in the JupyterLab, we see it's ordered by by the pickup time date here. It's not actually adhering to the date borders, but it's building up its own chunks because the data is not, we don't have the same number of taxi trips each day, but it's already sorted just by the number of trips per day. Does this guarantee that groups will, guys with the same index will be locally stored the same worker or will it still have a mixed workflow? I think it will have, it depends on, if you have just one single value it will be on worker. So you have to also be careful how you set your index. If you set all to the same value it will be all on the same worker. But in our case we don't have the same value because we're doing date times and not dates. And in that case, it's already an order data frame, and we can also look inside here. So this is now sorted, so head and tail actually make sense. We've got the 1st of January and the 31st of December here, and we're nearly to the new year at the end. And there's a thing called divisions and tasks, which is actually giving the minimum and maximum of the splits of a data frame, where we see that the first chunk of our data frame is on the first and the second. And then the second chunk is only on the third. And here you can see, when you're using the log indexer in Pandas, you actually can select just a chunk of your data or a chunk that matches your data on the index and get this chunk out locally. All you can do if you don't call compute, you can work with that chunk in memory going on and Dask will pass the operation to that. And also this case here, if you want to make a group operation, this case is called resample. If you call resample, it will change the dates to be on a one-hour basis and nothing more anymore behind that. And we can call mean on each of these groups. As we have ordered by pickup time, all the mean computations can already be done on a single node. So hopefully there will be no more communication. There was some communication. That's not what I hoped for. But yeah, there was some communication because we were not on date boundaries, not on hour boundaries. But if we already had our index set on hour boundaries up here, there wouldn't have been any communication. But also you can compare here, the communication here is quite small compared to this transfer here. It's still about only half of that, but that's already a big gain. Yeah. Is there a way to set your index before actually ingesting the data? Yeah, you can set if you're using parquet files. I think at the moment it's partly implemented for the backends, but you can already read out the index if you have an index in them. Or if you, a DAS data frame doesn't always have to come from a VCSV or read parquet, but it also can be created by yourself if you have just some efforts that produce data frames and up on them where you can build up a DAS data frame. Then you also can pass in the index. Or don't actually pass in the index, I think. You pass in the divisions. This here. And this gives DAS enough information so it knows how to distribute the work. so are there any more questions It's not the average of each of them. So you need to get the sum and then divide it. So this is done at the Dask load line. Yeah. So Dask knows itself when it has to do a mean, that it has to do summations on a local data frame, collect this, and then divide. That's the logic that Dask brings you. And if we have, like, a test on computation, do you apply it to the frame? Applying a DataFrame is basically what you're doing with MapPetitions, because in MapPetitions you get the real Pandas DataFrame you have in memory, and you can do anything you want on that. So normally if you want to do a DataFrame.Apply, you would write a function which gets a DF And then you can just work on this local data frame, maybe call shape, and then return the result again. And then my partitions will take this function, apply it on all data frames that are stored locally, and then you get the result out of that again. And then you can reduce whatever . So this is the section where you can use all your Pandas code logic. There's data.divisions. So data.divisions. For mine, actually, I have five alters, but it's many divisions. What's the difference between a division and a partition? I think partition and division in this case are the same things. So you have one big DAS data frame, but this one is split up into chunks like we told DAS on loading it, or DAS does some estimations when it loads things. If it's reading CSV files, I think it's doing chunks of 128 megabytes, and that will be a division at the end, or a chunk, or partition. All these words apply to the same things, and it does this independently of how many workers it has. But the number of partitions is the maximum number of parallel workers you can use for a computation. Yeah. OK? Good. So any more questions to data frames? Yeah. That's because there's, the question was, I'm repeating this for the video, when I call head or tail on a data frame, why can I call it without compute? There are some special computations or function calls in Das DataFrame, which actually already call compute behind the scenes. They also have a parameter, I think Executor or so, which you can pass then your client in, but normally we'll use the global client, because there are some things where it makes no sense to make them delayed, because the result they return is not something you will use in a distributed fashion. I think the same thing is doing a call shape. No shape is even making sense, but there are other things. I think if you call a group by a circle also, run immediately. So the next thing with Dask is going a layer deeper. Dask DataFrame is a convenient layer. If everything you do matches the DataFrame and also matches Dask DataFrame, then you're kind of already happy. but from experience I can tell you even if you're working with data frames not everything is working with Dask data frame because it's a high level API and sometimes it does not provide the performance you need or it's just too dumb so that you can't tell it what it actually should do. In this case there is the low level structure of Dask which is called Dask Delayed which is similar to Python future. And there's also a Python future API available. And this is the one function you can use to build up task graphs. There's also a link in here where you get a more complicated example how to do credit risk modeling with Dask. We're just doing a lot with Dask delayed objects because it's complicated. But we're going to do here a kind of simple use case. because this is also then useful if you just have really tasks that you want to scatter in parallel with some dependencies which are not data frames and are not arrays. In our simple case, we have two functions, an increment function and an add function, and we also added here some kind of magic work so they're not immediately done. Directly in our Python cluster, we can call them function, because there is some magic work inside, we have three functions and they take three seconds of time to execute. Now, as you can see in these functions, x and y are independent and are going into set, so actually we can compute x and y at the same time. We don't have to wait for one or the other, which we were doing here in this function. That's where you can use task to build up a task graph. We make those functions lazy by wrapping them into a task delayed object. Task delayed means you have to keep hold of the function and you can't pass in parameters. And the delayed object that's coming out of that will actually keep track of the passing parameters. But it won't do anything now. As you can see, calling these functions now takes 600 or 700 microseconds, which is not even enough time to do any amount of work. But we get a task graph out there. It's called set here. We can call compute. And we get the same result as above, a five. But this time it just took two seconds. And when we look at that, what is set actually? What have we computed? It's just a delayed object with a UUID behind it. Meaning that set already knows its dependencies, knows everything, but all that is packed into the delayed object. We can also look at the task in terms of what I just tried earlier, which wasn't a good idea with a 3000 node graph, but in this case it's a 3 node graph. We see we have two inputs, the two INC functions. They get each a result, which is then passed to the add function that also has a result. And as you also can see in this graph, these two functions are independent of each other. So Dask already can, if it has a multi-core scheduler, can run both in parallel, and only add has to wait for the result of both. Also, if you're kind of like me and don't like disappearing task graphs from top to down, there's this hidden parameter called RankDir, which is a graphless parameter where you can just flip the graph and then you have it from left to right, your data flow, which is sometimes more easy to view. there's some more questions you can ask if you want to see if you have understood it but actually now there's also a small exercise here which you also can do now by your own it's a simple loop here which you can run in sql and yeah sequential not in sql and see if you by your own skills can change it so that in the end it's a delayed object. And you can actually parallelize the loop. Okay, I think most people are already through it. The solution here is mainly you wrap things in dust delayed objects and you run it but at the end it takes 2 milliseconds simply because you need to call compute and then it really runs in 2 seconds mainly because it needs 1 second for all those ink computations they all run in parallel and 1 second I'm confused now I'm using some and not add. Okay, now the number of cores I have is limited, so I can't run all in parallel. That means I have two seconds because my local cluster I'm running in, this is not a distributed cluster, only has probably four cores. So each worker is executing two ink functions, meaning each worker has two seconds of work and then the summation at the end is summation over all the distillate results they can happen in parallel and it takes only two seconds there's also a different way you can approach this you can just call das compute on an array and then can use the normal sum function on them If you use task computer and array, and if you use it without a time at the top, you get the real direct results. And then you can use sum over the results at the end. And this gives you the result directly, but it still leaves you with an intermediate array. If you have a bunch of delayed objects and now find a reducer at the end, or just want a list as a result at the end, last compute calls compute on all these single delayed objects, and then waits until all of them are computed. Yeah. Yes. So that's also a thing we can just do here. If we just do some results, this also returns a delayed object. That's basically because the plus operator is overloaded on these delayed objects. That's also here, which you can see in the visualization. If you call sum, you're always doing pairwise aggregation. So you have your initial first increment call that's finished. Then it takes a second one. You do plus, and it gives you a new delayed object. And it just goes up a summation here. This looks like a dev first search. And if you're doing a delayed sum, it will take all ink results in at the same time. So, because time has already progressed a bit, I'm going to skip the step on the bottom part here, which tells you how you can actually build up, using DAST delayed, custom data frame algorithms. Basically, you will start off loading single files into memory, wrapping this single file loading into DasDelayed objects, and you can work with these DasDelayed objects, each on them, and at the end call DasDataFrameFromDelayed. But another thing which you also really could do good with Dask and Dask-Delayed is doing machine learning. And you don't even need a complex model which is trying to train a huge amount of data distributed over your cluster. But the simplest thing is you have a simple scikit-learn example, in this case the newsgroup example, which tries to classify some text into categories. This data all fits into the RAM of your machine, and you can also just work on your local machine in your laptop, run a simple pipeline that splits up the text, does some TF-IDF, and then a classifier at the end, and everything works fine. RAM on your laptop is sufficient. The thing that's not sufficient on your laptop is if you're doing a grid search on the parameters. That will take a long time, and every parameter you add to your grid search will double the time at least, or even more, because it has a bigger grid to search. That's the thing where you can actually then take the ask, do really trivial parallelization, and say, okay, I have this big graph, and I want to train all these grid search models in parallel. If you are approaching this on a dump way, you take your grid search, you unroll the for loops, which you take your main training data in here, make a for loop unrolling like like we had this grid CV up there. And in the lowest for loop, we call std classifier, which is the last step of our pipeline. And that's basically what's happening up here with this simple scikit-learn pipeline and doing grid search. Now, instead of doing that in this really complicated for loop, you can actually start and pass this out to your cluster and using delayed objects. Yeah, I think I have nothing in there which actually shows it, but this is a good example. You can actually do it on your laptop or you can just think of it now because it's a bit more compute intensive has already someone an idea what we should do here to get it fast and in parallel can explain a bit more it starts up here it takes the data grant account vectorizer on it or it creates a convict rise that runs fit runs transform on training and test data and then returns an STD classifier. And also it fits it again, scores it, and at the end you have scores. And you add these scores into a big list. Yes. yes and so you can repeat that we you can wrap the vectorizer and a classifier here to fit and the score methods into delayed objects use the delayed results but also what you can do here is these are nested for looks so up here you will just take the data wrap that into the date the main thing is if you would read the data now into your local local notebook it will be in your notebook. And if you start a computation on the cluster, the first thing that will happen is it will transfer the data from your notebook to the cluster. If you wrap it into a delayed object, the worker, which is starting with the computation, will pull it into memory, but it will pull it from storage into the worker. If it's in your notebook, it gets pickled and sent over to the cluster. If it's a complicated data structure, pickling will take a lot of time. And if you're like me here at PyCon, and your cluster is somewhere in the Google Cloud in the it will take also a long way with a big amount of data. So also if you're doing any I.O., wrap it into the delayed objects right in the cluster. Then you have a count vectorizer, which is in your loop. You will generate several delayed objects, each instantiation of that. But also down here, you have another two for loops, and you run the classifier on that. And you will make one delayed object up here, where you just fit a count vectorizer, and then you can reuse it down here in this case just four times because we only have a matrix of four but you only have to fit a counter-rectorizer once and can use it for four classifier steps which will not make it just a linear task graph but actually will be a task graph which fans out a bit later on and then all fans in back again into the parameter scores list. And this is kind of a simple example or the code is a bit longer although it's more complicated, but it's a simple way how you can already parallelize machine learning code with Dask. This one here is now kind of a self-made parallelization, but also if you just look at parallelizing machine learning, there is a Dask ML library which already provides some scikit-learn-based stuff and some other things, which actually gives you already this parallelization which we've written by hand here. So the sentence at the desk delay goes on to be How does it handle things like attribute errors? Like if you're calling fit on a delayed object, do you only get the error after it's used though? Or is it doing it later, like it's not checking, or? It can't check really because the functions aren't executed. So attribute errors, or general errors, can only happen on compute. But there's also a nice thing we have to ask in here. So we can just do some func again, and this time with no argument, but raise not implemented error. And you see, when you just make the delayed object, you have now a delayed function. You call it, it's still lazy, and only on the moment where you call compute, you will get a not implemented error. One of the nice things is it also propagates the error to your machine. So if you have an error on your cluster somewhere, it will be on your machine. And there's even a nice add-on in Dask, which I have never used, but it's probably going to be really useful, is you can get a debugger in your notebook on your worker, which is somewhere else. But if there are some failures, you can just have a look into that. So now, the workshop has only 50 minutes remaining. Because it's a workshop and you can ask questions, I suggest mostly to you just look again at some of those examples. Try these examples. Run them. Ask questions. overall i hope this was a good introduction for you to to get into dask if it was really low level really simple but i think if you never have dealt with dask before and this should give you a rough overview what's there also dask arrays work similar to the dask um and data frame api you you get it and a task array instance which is made up it's mostly a numpy matrix on hundreds of NumPy matrix in the back end. They all chunk across your cluster. You can call map partitions. I'm not sure if it's called map partitions or map divisions on arrays, where you can do your custom NumPy operations. But there's also some high level operations based on these arrays where you can just call sum or FFT or so. And Das Back is the other one, which gives you distributed collections, which is a bit more simple. So it gives you mostly a possibility to read something into Dask, which is a list or so at the end, or a distributed list. And then you get simple functions like map or read use. And the good thing is, if you have just one of these high-level data structures in Dask, you can always call toDelayed, and you get a list of delayed objects, which you then also can use to just build these low-level delayed functions and do your custom stuff. So if Dask is not providing what you need at the moment in high-level API, or it's not working as you would expect and you know how to write it in a distributed fashion, then you can always convert to DL8, write your own custom algorithm, and once you're done or you can go back to the Dask APIs, all those Dask high-level functions have a fromDelayed, and then you can continue with the high-level Dask stuff. If you have anything that's hard to pickle, make a function that can create it and only pass around the function that can create the instance. So if you have a database handle, make a function that gets a config with it, and the function in the config can get pickled, and then the functions in the generator will always generate a connection to the database. But don't pass around IO connections. Even if you have a network connection with requests or so, always just pass around the things you need, the URL and maybe the credentials, as a pickled object, and only create it if needed. But also, the important thing is, look that your arguments you pass always to delayed functions are small. Everything which is in your notebook gets transferred over the cluster. If you make it too complicated, your computation will be slow just because of the scheduling and the pickling of the code. So normally when I'm using Dask, I only have configuration YAML or so on the scheduler which is running, everything else like loading even medium sized files or small sized files is done on the cluster. But that basically was the tutorial, I'm still here around at least until half past to answer questions but I'm not going to show anything more except there are questions where I can show something also to do some advertisement for talks this is the first tutorial I was using using binder it worked really well and if you want to know how binders working what is doing there is a talk in one hour in I think in Kubus. You mean the setup with the desk things on the right? There is a configuration in JupyterLab, but you have to look into the repo there. That's the thing why I talked from Matthew Rocklin, because Matthew made a setup. My setup was just having notebooks. My old presentation was using notebooks and using the presentation mode of Jupyter, which is also nice, but the Jupyter extension at the side, which you can access here, is much, much prettier. Does Binary use JupyterLab by default, or do you have to characterize your samples? I think that's the question you can ask in the alpha talk. I have no idea. Do you know how to export the ? Yeah, you can just go here on the folder, and then right click, and there's a download. OK. OK. Sorry. Yeah. How do you keep your code? Is your own library specific to PIP? At work, we're using Apache Mesos and Aurora. And when we're starting a DAS class with Aurora, we are also telling it which libraries it should have. And as we're using PIP, we're giving it an exact pinning of all libraries used. Yeah. So that's when you're restarting a new worker and you're just doing PIP install, it will actually install all the exact versions, even if there's a new version of a dependency in there. I have some colleagues that have a workaround that there's code in Dask which you can specify that should run on each worker startup. If you're in an interactive setting and not productive, you can actually add that code which does a pip install. So if you're just working in an interactive setting and you forgot to install a package, you can use that hack to add new packages in there. But the basic thing is, I always have a pin set of requirements installed on all my worker nodes and in my Jupyter lab. And so everything is consistent. Because once things get inconsistent, you get to get weird errors, because Python versions don't match anymore.