- Dask Cookbook
-
-
-
This Project Pythia Cookbook provides a comprehensive guide to understanding the basic concepts and collections of Dask as well as its integration with Xarray. -Dask is a parallel computing library that allows you to scale your computations to multiple cores or even clusters, while Xarray is a library that enables working with labelled multi-dimensional arrays, with a focus on working with netCDF datasets.
-Motivation
-The motivation behind this repository is to provide a clear and concise resource for anyone looking to learn about the basic concepts of Dask and its integration with Xarray. By providing step-by-step tutorials, we hope to make it easy for users to understand the fundamental concepts of parallel computing and distributed data processing, as well as how to apply them in practice using Dask and Dask+Xarray.
-Authors
-Negin Sobhani, Brian Vanderwende, Deepak Cherian, and Ben Kirk
-Contributors
- -Note on Content Origin
-This cookbook is derived from the extensive material used in the NCAR tutorial, “Using Dask on HPC systems”, which was held in February 2023. The NCAR tutorial series also includes an in-depth exploration and practical use cases of Dask on HPC systems and best practices for Dask on HPC. For the complete set of NCAR tutorial materials, including these additional insights -on Dask on HPC, please refer to the main NCAR tutorial content available here.
-Structure
-In the first chapter of this cookbook, we provide step-by-step tutorials on the basic concepts of Dask, including Dask arrays and Dask dataframes, which are powerful tools for parallel computing and distributed data processing. We explain the key differences between these Dask data structures and their counterparts in NumPy and Pandas.
-In the second chapter of the repository, we move on to more advanced topics, such as distributed computing and Dask+Xarray integration. We provide examples of how to use Dask+Xarray to efficiently work with large, labelled multi-dimensional datasets. -Finally, we will discuss some best practices regarding Dask+Xarray.
-Running the Notebooks
-You can either run the notebook using Binder or on your local machine.
-Running on Binder
-The simplest way to interact with a Jupyter Notebook is through -Binder, which enables the execution of a -Jupyter Book in the cloud. The details of how this works are not -important for now. All you need to know is how to launch a Pythia -Cookbooks chapter via Binder. Simply navigate your mouse to -the top right corner of the book chapter you are viewing and click -on the rocket ship icon, (see figure below), and be sure to select -“launch Binder”. After a moment you should be presented with a -notebook that you can interact with. I.e. you’ll be able to execute -and even change the example programs. You’ll see that the code cells -have no output at first, until you execute them by pressing -Shift+Enter. Complete details on how to interact with -a live Jupyter notebook are described in Getting Started with -Jupyter.
-Running on Your Own Machine
-If you are interested in running this material locally on your computer, you will need to follow this workflow:
--
-
Clone the
-https://github.com/ProjectPythia/dask-cookbookrepository:--git clone https://github.com/ProjectPythia/dask-cookbook.git -
-Move into the
-dask-cookbookdirectory--cd dask-cookbook -
-Create and activate your conda environment from the
-environment.ymlfile--conda env create -f environment.yml -conda activate dask-cookbook -
-Move into the
-notebooksdirectory and start up Jupyterlab--cd notebooks/ -jupyter lab -
-
Acknowledgments
--
-
NCAR CISL/CSG Team
-ESDS Initiative
-
![\"Dask](\"https://docs.dask.org/en/stable/_images/dask_horizontal.svg\"\n",)
\n",
- "\n",
- "# Dask Overview\n",
- "\n",
- "### In this tutorial, you learn:\n",
- "\n",
- "* What is Dask?\n",
- "* Why Dask in Geosciences?\n",
- "* Dask Data Structures and Schedulers\n",
- "* When to use Dask?\n"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "a26e012c-213f-43ad-997c-1f895241826e",
- "metadata": {
- "tags": []
- },
- "source": [
- "## Introduction\n",
- "\n",
- "Complex data structures enable data science in Python. For example:\n",
- "* [NumPy arrays](https://numpy.org/doc/stable/)\n",
- "* [Pandas series and dataframes](https://pandas.pydata.org/)\n",
- "* [XArray datasets](https://docs.xarray.dev/)\n",
- "\n",
- "*But datasets are getting larger all of the time! What if my dataset is too big to fit into memory, or it takes too long to complete an analysis?*\n",
- "\n",
- "## What is Dask?\n",
- "\n",
- "* Dask is an open-source Python library for parallel and distributed computing that scales the existing Python ecosystem.\n",
- "\n",
- "* Dask was developed to scale Python packages such as Numpy, Pandas, and Xarray to multi-core machines and distributed clusters when datasets exceed memory.\n",
- "\n",
- "\n",
- "## Why Dask?\n",
- "\n",
- "### Familiar Interface \n",
- "\n",
- "Dask provides interfaces which mimics significant portions of the NumPy and Pandas APIs. \n",
- "\n",
- "This means Dask provides ways to parallelize Pandas, Xarray, and Numpy workflows with minimal code rewriting (no massive code-restructure or writing a script in another language).\n",
- "\n",
- "### Scalability\n",
- "Dask is designed to scale well from single machine (laptop) to thousand-node HPC clusters, and on the cloud.\n",
- "\n",
- "This allows users to use their existing hardware, or add more machines as needed, to handle increasingly large and complex datasets.\n",
- "\n",
- "### Flexibility\n",
- "Dask provides several tools that help with data analysis on large datasets. For example, you can easily wrap your function in `dask.delayed` decorator to make it run in parallel. \n",
- "\n",
- "Dask provides seamless integration with well-known HPC resource managers and job scheduling systems, including PBS, SLURM, and SGE.\n",
- "\n",
- "### Built-in Diagnostic Tools\n",
- "Dask provides responsive feedback via the client as well as a real-time interactive diagnostic dashboard to keep users informed on how the computation is progressing. \n",
- "\n",
- "This helps users identify and resolve potential issues without waiting for the work to be completed. \n",
- "\n",
- "\n",
- "## First Rule of Dask\n",
- "\n",
- "While Dask is a powerful tool for parallel and distributed computing, it is not always the best solution for every problem. \n",
- "In some cases, using Dask may introduce additional complexity and overhead, without providing any substantial benefits in terms of performance or scalability.\n",
- "\n",
- "* Keep in mind the time spent parallelizing and optimizing your workflow when using Dask vs. the time saved because of that parallelization.\n",
- "\n",
- "* Consider how many times you plan to run your code - if only once, is it worth it?\n",
- "\n",
- "![](\"https://raw.githubusercontent.com/ProjectPythia/dask-cookbook/main/notebooks/images/dask_twitter.png\"\n",)
\n",
- "\n",
- "![\"Dask](\"https://raw.githubusercontent.com/ProjectPythia/dask-cookbook/main/notebooks/images/high_vs_low_level_coll_analogy.png\"\n",)
\n",
- " \n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "\n",
- "### 2. Dynamic Task Scheduling\n",
- "**We can basically think of the Dask scheduler as our task orchestrator.**\n",
- "\n",
- "When a computation is submitted, work is segmented into discrete tasks which are assigned to workers by the Dask scheduler. \n",
- "\n",
- "To perform work, a scheduler must be assigned resources in the form of a Dask cluster. The cluster consists of the following components: \n",
- "\n",
- "* **scheduler** : A scheduler creates and manages task graphs and distributes tasks to workers.\n",
- "\n",
- "* **workers** : A worker is typically a separate Python process on either the local host or a remote machine. A Dask cluster usually consists of many workers. Basically, a worker is a Python interpretor which will perform work on a subset of our dataset.\n",
- "\n",
- "* **client** - A high-level interface that points to the scheduler (often local but not always). A client serves as the entry point for interacting with a Dask scheduler.\n",
- "\n",
- "![\"Dask](\"https://tutorial.dask.org/_images/distributed-overview.png\"\n",)
\n",
- " \n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "\n",
- "We will learn more about Dask Collections and Dynamic Task Scheduling in the next tutorials."
- ]
- },
- {
- "cell_type": "markdown",
- "id": "9cbea0a3-6f90-41a1-9827-8f60771e67af",
- "metadata": {},
- "source": [
- "## Useful Resources\n",
- "\n",
- "* Reference\n",
- " * [Docs](https://dask.org/)\n",
- " * [Examples](https://examples.dask.org/)\n",
- " * [Code](https://github.com/dask/dask/)\n",
- " * [Blog](https://blog.dask.org/)\n",
- "* Ask for help\n",
- " * [`dask`](http://stackoverflow.com/questions/tagged/dask) tag on Stack Overflow, for usage questions\n",
- " * [github issues](https://github.com/dask/dask/issues/new) for bug reports and feature requests\n",
- " * [discourse forum](https://dask.discourse.group/) for general, non-bug, questions and discussion"
- ]
- }
- ],
- "metadata": {
- "kernelspec": {
- "display_name": "Python 3 (ipykernel)",
- "language": "python",
- "name": "python3"
- },
- "language_info": {
- "codemirror_mode": {
- "name": "ipython",
- "version": 3
- },
- "file_extension": ".py",
- "mimetype": "text/x-python",
- "name": "python",
- "nbconvert_exporter": "python",
- "pygments_lexer": "ipython3",
- "version": "3.9.18"
- },
- "widgets": {
- "application/vnd.jupyter.widget-state+json": {
- "state": {},
- "version_major": 2,
- "version_minor": 0
- }
- }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}
diff --git a/_preview/22/_sources/notebooks/01-dask-array.ipynb b/_preview/22/_sources/notebooks/01-dask-array.ipynb
deleted file mode 100644
index fc7d2bf..0000000
--- a/_preview/22/_sources/notebooks/01-dask-array.ipynb
+++ /dev/null
@@ -1,1121 +0,0 @@
-{
- "cells": [
- {
- "cell_type": "markdown",
- "id": "990cfa4c-2117-4435-9806-ff9048890398",
- "metadata": {
- "tags": []
- },
- "source": [
- "![](\"https://docs.dask.org/en/stable/_images/dask-array.svg\")
\n",
- "\n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "Dask Array can be used as a drop-in replacement for NumPy arrays, with a similar API and support for a subset of NumPy functions. \n",
- "\n",
- "Dask effectively reduces the memory footprint of large array computations by dividing the arrays into smaller pieces (called **chunks**) that can fit into memory and stream the data from disk."
- ]
- },
- {
- "cell_type": "markdown",
- "id": "106238d9-c519-47bc-b5c6-5c703ad47fe6",
- "metadata": {},
- "source": [
- "**Dask Arrays are lazy:** Unlike Numpy, operations on Dask arrays are not computed until you explicitly request them. \n",
- "\n",
- "![](\"https://d33wubrfki0l68.cloudfront.net/f7bf6ca40c8f217386f83795b36e0c964c6a9d2b/ad6da/images/blog/what-is-dask-blockwise-sum.jpg\")
\n",
- "\n",
- "*Image adapted from saturncloud.io*\n"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "af956be7-75fe-4d95-8212-76b8fabf1a1a",
- "metadata": {},
- "source": [
- "## Task Graph\n",
- "\n",
- "The Dask Task Graph serves as a **blueprint** for executing the computations. \n",
- "\n",
- "The Task Graph defines the (1) relationships between tasks, and (2) the order in which they should be executed.\n",
- "\n",
- "In a task graph each node in the graph represents a task and lines represent the dependencies/relationships between tasks."
- ]
- },
- {
- "cell_type": "markdown",
- "id": "0fd27324-4077-46a4-97d9-6a01a61ab61d",
- "metadata": {},
- "source": [
- "We can visualize the low-level task graph using `.visualize()` method."
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "32303572-b72a-49e2-9a59-165d11e2da1c",
- "metadata": {},
- "outputs": [],
- "source": [
- "darr.sum().visualize(rankdir=\"LR\")"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "9cf9a3cb-9187-405d-a176-9530ab64cdd7",
- "metadata": {},
- "source": [
- "It is generally good practice to look at the task graph before executing the computation. By looking at the task graph, you can learn about potential bottlenecks where parallelism is not possible. \n",
- "\n",
- "![\"Dask](\"https://docs.dask.org/en/stable/_images/dask-dataframe.svg\"\n",)
\n",
- "\n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "pandas is a very popular tool for working with **tabular datasets**, but the dataset needs to **fit into the memory**. \n",
- "\n",
- "pandas operates best with smaller datasets, and if you have a large dataset, you’ll receive an out of memory error using pandas. A general rule of thumb for pandas is:\n",
- "\n",
- "> \"Have 5 to 10 times as much RAM as the size of your dataset\"\n",
- ">\n",
- "> ~ Wes McKinney (2017) in [10 things I hate about pandas](https://wesmckinney.com/blog/apache-arrow-pandas-internals/)\n",
- "\n",
- "But Dask DataFrame can be used to solve pandas performance issues with larger-than-memory datasets.\n",
- "\n",
- "\n",
- "### What is Dask DataFrame?\n",
- "\n",
- "\n",
- "* **A Dask DataFrame is a parallel DataFrame composed of smaller pandas DataFrames (also known as *partitions*).**\n",
- "\n",
- "* Dask Dataframes look and feel like the pandas DataFrames on the surface. \n",
- "\n",
- "* Dask DataFrames partition the data into manageable **partitions** that can be processed in parallel and across multiple cores or computers. \n",
- "\n",
- "* Similar to Dask Arrays, Dask DataFrames are lazy!\n",
- "\n",
- " Unlike pandas, operations on Dask DataFrames are not computed until you explicitly request them (e.g. by calling `.compute`).\n",
- "\n"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "bae50738-0e67-4a64-8190-12ab8c06e06c",
- "metadata": {},
- "source": [
- "## When to use Dask DataFrame and when to avoid it?\n",
- "\n",
- "Dask DataFrames are used in situations where pandas *fails* or has *poor performance due to data size*. \n",
- "\n",
- "Dask DataFrame is a good choice when doing **parallalizeable computations**. \n",
- "Some examples are:\n",
- "* Element-wise operations such as `df.x + df.y`\n",
- "* Row-wise filtering such as `df[df.x>0]`\n",
- "* Common aggregations such as `df.x.max()`\n",
- "* Dropping duplicates such as `df.x.drop_duplicate()`\n",
- "\n",
- "However, Dask is not great for operations that requires shuffling or re-indexing. \n",
- "Some examples are:\n",
- "* Set index: `df.set_index(df.x)`\n",
- "\n",
- "\n",
- " \n",
- "![\"News](\"https://raw.githubusercontent.com/ProjectPythia/dask-cookbook/main/notebooks/images/denver.png\"\n",)
\n",
- "\n",
- "Below, we show all days with high temperature above 60°F (155.5°C/10) since 2010:"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "29c25614-b47c-4baa-b429-8e3f17e3a958",
- "metadata": {},
- "outputs": [],
- "source": [
- "df[(df['MONTH']==1) & (df['YEAR']>=2010) & (df['TMAX']>155.5)].groupby(['YEAR']).size()"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "86eb54cb-9358-4ef9-b913-a8f653993ae5",
- "metadata": {},
- "source": [
- "This is great! But how big is this dataset for one station?\n",
- "\n",
- "First, let's check the file size:"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "eb041314-2253-4933-ab2c-c81f742f44fc",
- "metadata": {},
- "outputs": [],
- "source": [
- "!ls -lh ../data/USW00003017.csv"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "5c84f8f2-94f1-45d0-b789-82c779aa3ce0",
- "metadata": {},
- "source": [
- "Similar to the previous tutorial, we can use the following function to find the size of a variable on memory."
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "de8d54fc-ec05-41c1-a624-274e40772d13",
- "metadata": {},
- "outputs": [],
- "source": [
- "# Define function to display variable size in MB\n",
- "import sys\n",
- "def var_size(in_var):\n",
- " result = sys.getsizeof(in_var) / 1e6\n",
- " print(f\"Size of variable: {result:.2f} MB\")"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "16c0b151-1454-4e38-96c6-41c38434b9cb",
- "metadata": {},
- "outputs": [],
- "source": [
- "var_size(df)"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "2cae6ce6-0dca-47f1-bf84-ca3b46bfbb91",
- "metadata": {},
- "source": [
- "Remember, the above rule?\n",
- "\n",
- "> \"Have 5 to 10 times as much RAM as the size of your dataset\"\n",
- ">\n",
- "> ~ Wes McKinney (2017) in [10 things I hate about pandas](https://wesmckinney.com/blog/apache-arrow-pandas-internals/)\n"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "ccf0ba8b-8c65-40ce-81cb-0c313177fc8b",
- "metadata": {},
- "source": [
- "**So far, we read in and analyzed data for one station. We have a total of +118,000 stations over the world and +4500 stations in Colorado alone!**\n",
- "\n",
- "**What if we want to look at the larger dataset?**\n",
- "\n",
- "## Scaling up to a larger dataset\n",
- "\n",
- "Let's start by reading data from selected stations. The downloaded data for this example includes the climatology observations from 66 selected sites in Colorado.\n",
- "\n",
- "Pandas can concatenate data to load data spread across multiple files:\n"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "25e0edd0-a58a-4ad4-8ab7-551122c92d14",
- "metadata": {},
- "outputs": [],
- "source": [
- "!du -csh ../data/*.csv |tail -n1"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "d1258cc3-d24a-4268-ad19-62e02905b635",
- "metadata": {},
- "source": [
- "Using a for loop with `pandas.concat`, we can read multiple files at the same time:"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "ddfc9c27-00da-426d-bb45-83f959f411ec",
- "metadata": {},
- "outputs": [],
- "source": [
- "%%time\n",
- "import glob\n",
- "co_sites = glob.glob(os.path.join(data_dir, '*.csv'))\n",
- "df = pd.concat(pd.read_csv(f, index_col=0, parse_dates=['DATE']) for f in co_sites)"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "3d16de06-b7f5-4d4a-ac71-011d7ce5e1b1",
- "metadata": {},
- "source": [
- "* How many stations have we read in?"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "4441c14e-a51b-4f59-b7e9-2fa0bdb1f6eb",
- "metadata": {},
- "outputs": [],
- "source": [
- "print (\"Concatenated data for\", len(df.ID.unique()), \"unique sites.\")"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "9ab38e3b-1c37-48f8-86ca-903192cfcc1f",
- "metadata": {},
- "source": [
- "Now that we concatenated the data for all sites in one DataFrame, we can do similar analysis on it:"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "e8fee981-ef7c-4bbb-b7a2-db4705b3902e",
- "metadata": {},
- "source": [
- "* Which site has recorded the most snow days in a year?"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "4509a4e3-ab07-4bd5-bbb8-fe121e481891",
- "metadata": {},
- "outputs": [],
- "source": [
- "%%time\n",
- "# ~90s on 4GB RAM\n",
- "snowy_days = df[df['SNOW']>0].groupby(['ID','YEAR']).size()\n",
- "\n",
- "print ('This site has the highest number of snow days in a year : ')\n",
- "snowy_days.agg(['idxmax','max'])"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "160a297c-96cb-4014-a196-72e1482901a9",
- "metadata": {},
- "source": [
- "**Excersise:** Which Colorado site has recorded the most snow days in 2023?"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "1bbd5656-49a3-448c-b339-c6844304f61f",
- "metadata": {},
- "source": [
- "**Dask allows us to conceptualize all of these files as a single dataframe!**"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "63aff343-4431-4224-9c82-9ed725c9ea0c",
- "metadata": {},
- "outputs": [],
- "source": [
- "# Let's do a little cleanup\n",
- "del df, snowy_days"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "3746273b-4057-466d-b7c9-a78932cd5808",
- "metadata": {},
- "source": [
- "## Computations on Dask DataFrame"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "6df1339a-7ee4-4125-b281-c467358d79a5",
- "metadata": {},
- "source": [
- "### Create a \"LocalCluster\" Client with Dask"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "bb859ca6-a98d-40cd-a389-289c5246f670",
- "metadata": {},
- "outputs": [],
- "source": [
- "from dask.distributed import Client, LocalCluster\n",
- "\n",
- "cluster = LocalCluster()\n",
- "client = Client(cluster)\n",
- "client"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "db5ae984-5763-4360-ab1b-2ecf92500952",
- "metadata": {},
- "source": [
- "☝️ Click the Dashboard link above.\n",
- "\n",
- "👈 Or click the “Search” 🔍 button in the dask-labextension dashboard.\n",
- "\n",
- "### Dask DataFrame `read_csv` to read multiple files\n",
- "\n",
- "`dask.dataframe.read_csv` function can be used in conjunction with `glob` to read multiple csv files at the same time. \n",
- "\n",
- "Remember we can read one file with `pandas.read_csv`. For reading multiple files with pandas, we have to concatenate them with `pd.concatenate`. However, we can read many files at once just using `dask.dataframe.read_csv`.\n",
- "\n",
- "Overall, Dask is designed to perform I/O in parallel and is more performant than pandas for operations with multiple files or large files."
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "6d63add1-7e7f-4631-ae82-19800f4122a5",
- "metadata": {},
- "outputs": [],
- "source": [
- "%%time\n",
- "import dask\n",
- "import dask.dataframe as dd\n",
- "\n",
- "ddf = dd.read_csv(co_sites, parse_dates=['DATE'])\n",
- "ddf"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "c8147b02-8f8d-4591-92c7-be3441bea8ae",
- "metadata": {},
- "outputs": [],
- "source": [
- "ddf.TMAX.mean()"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "515a8028-c8f0-4bcf-8841-d4506fafb9d7",
- "metadata": {},
- "source": [
- "**Notice that the representation of the DataFrame object contains no data just headers and datatypes. Why?**\n",
- "\n",
- "#### Lazy Evaluation\n",
- "\n",
- "Similar to Dask Arrays, Dask DataFrames are lazy. Here the data has not yet been read into the dataframe yet (a.k.a. lazy evaluation). \n",
- "Dask just construct the task graph of the computation but it will \"evaluate\" them only when necessary.\n",
- "\n",
- "**So how does Dask know the name and dtype of each column?**\n",
- "\n",
- "Dask has just read the start of the first file and infers the column names and dtypes. \n",
- "\n",
- "Unlike `pandas.read_csv` that reads in all files before inferring data types, `dask.dataframe.read_csv` only reads in a sample from the beginning of the file (or first file if using a glob). The column names and dtypes are then enforced when reading the specific partitions (*Dask can make mistakes on these inferences if there is missing or misleading data in the early rows*)."
- ]
- },
- {
- "cell_type": "markdown",
- "id": "f3f062ee-ca69-4262-8316-807705a8fa07",
- "metadata": {},
- "source": [
- "Let's take a look at the start of our dataframe:"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "f7b4823d-a785-495d-b2f5-e09a539f48d1",
- "metadata": {},
- "outputs": [],
- "source": [
- "ddf.head()"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "b4bb8ab3-444e-41e3-94a0-8ec1b6354751",
- "metadata": {},
- "source": [
- "![\"Xarray\"\n",](\"https://docs.xarray.dev/en/latest/_images/dataset-diagram.png\"\n",)
\n",
- "\n",
- "# Parallelizing Xarray with Dask\n",
- "\n",
- "### In this tutorial, you learn:\n",
- "\n",
- "* Using Dask with Xarray\n",
- "* Read/write netCDF files with Dask\n",
- "* Dask backed Xarray objects and operations\n",
- "* Extract Dask arrays from Xarray objects and use Dask array directly.\n",
- "* Xarray built-in operations can transparently use dask\n",
- "\n",
- "### Prerequisites\n",
- "| Concepts | Importance | Notes |\n",
- "| --- | --- | --- |\n",
- "| [Intro to Xarray](https://foundations.projectpythia.org/core/xarray/xarray-intro.html) | Necessary | |\n",
- "| Dask Arrays | Necessary | |\n",
- "| Dask DataFrames | Necessary | |\n",
- "\n",
- "- **Time to learn**: 40 minutes\n",
- "---------\n",
- "\n",
- "## Introduction\n",
- "\n",
- "### Xarray Quick Overview\n",
- "\n",
- "\n",
- " \n",
- "Xarray is an open-source Python library designed for working with *labelled multi-dimensional* data. By *multi-dimensional* data (also often called *N-dimensional*), we mean data that has many independent dimensions or axes (e.g. latitude, longitude, time). By labelled we mean that these axes or dimensions are associated with coordinate names (like \"latitude\") and coordinate labels like \"30 degrees North\".\n",
- "\n",
- "Xarray provides pandas-level convenience for working with this type of data.\n",
- "\n",
- "\n",
- "![\"High](\"https://docs.dask.org/en/stable/_images/dask-overview.svg\")
\n",
- "\n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "### 1. Single machine scheduler \n",
- "* The Single-machine Scheduler schedules tasks and manages the execution of those tasks on the same machine where the scheduler is running. \n",
- "* It is designed to be used in situations where the amount of data or the computational requirements are too large for a single process to handle, but not large enough to warrant the use of a cluster of machines.\n",
- "* It is relatively simple and cheap to use but it does not scale as it only runs on a single machine. \n",
- "\n",
- "**Single machine scheduler is the default choice used by Dask.**\n",
- "\n",
- "In Dask, there are several types of single machine schedulers that can be used to schedule computations on a single machine:\n",
- "#### 1.1. Single-threaded scheduler\n",
- "This scheduler runs all tasks **serially** on a single thread. \n",
- "This is only useful for debugging and profiling, but does not have any parallelization. \n",
- "\n",
- "#### 1.2. Threaded scheduler\n",
- "The threaded scheduler uses a pool of **local** threads to execute tasks concurrently. \n",
- "This is the default scheduler for Dask, and is suitable for most use cases on a single machine. Multithreading works well for Dask Array and Dask DataFrame. \n",
- "\n",
- "To select one of the above scheduler for your computation, you can specify it when doing `.compute()`:\n",
- "\n",
- "For example: \n",
- "```python\n",
- "this.compute(scheduler=\"single-threaded\") # for debugging and profiling only\n",
- "```\n",
- "\n",
- "\n",
- "As mentioned above the *threaded scheduler* is the default scheduler in Dask. But you can set the default scheduler to Single-threaded or multi-processing by: \n",
- "\n",
- "```python\n",
- "import dask\n",
- "dask.config.set(scheduler='synchronous') # overwrite default with single-threaded scheduler\n",
- "```\n",
- "\n",
- "Multi-processing works well for pure Python code - *delayed* functions and operations on Dask Bags.\n",
- "\n",
- "Let's compare the performance of each of these single-machine schedulers:"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": 1,
- "id": "5ed871a9-45d4-4b0e-9f39-4dbb0b871c64",
- "metadata": {},
- "outputs": [],
- "source": [
- "import numpy as np\n",
- "import dask.array as da"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": 14,
- "id": "f61fb1fe-bbd7-4207-8ac6-8eb784d6710f",
- "metadata": {},
- "outputs": [
- {
- "name": "stdout",
- "output_type": "stream",
- "text": [
- "CPU times: user 14.9 s, sys: 1.32 s, total: 16.2 s\n",
- "Wall time: 16.1 s\n"
- ]
- },
- {
- "data": {
- "text/plain": [
- "array([ 9.99987393, 9.99942047, 10.00069322, ..., 9.99997333,\n",
- " 9.99945909, 10.00094973])"
- ]
- },
- "execution_count": 14,
- "metadata": {},
- "output_type": "execute_result"
- }
- ],
- "source": [
- "%%time\n",
- "## - numpy performance\n",
- "xn = np.random.normal(10, 0.1, size=(20_000, 20_000))\n",
- "yn = xn.mean(axis=0)\n",
- "yn"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": 15,
- "id": "af609dc9-9482-41e0-8a93-5cbcf096e376",
- "metadata": {},
- "outputs": [
- {
- "name": "stdout",
- "output_type": "stream",
- "text": [
- "CPU times: user 14.8 s, sys: 112 ms, total: 14.9 s\n",
- "Wall time: 3.83 s\n"
- ]
- },
- {
- "data": {
- "text/plain": [
- "array([ 9.99928454, 9.99968075, 10.00027327, ..., 10.00030439,\n",
- " 9.9999113 , 9.99947802])"
- ]
- },
- "execution_count": 15,
- "metadata": {},
- "output_type": "execute_result"
- }
- ],
- "source": [
- "%%time\n",
- "# -- dask array using the default\n",
- "xd = da.random.normal(10, 0.1, size=(20_000, 20_000), chunks=(2000, 2000))\n",
- "yd = xd.mean(axis=0)\n",
- "yd.compute()"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": 16,
- "id": "141053bf-942c-46be-b169-2709ca569451",
- "metadata": {},
- "outputs": [
- {
- "name": "stdout",
- "output_type": "stream",
- "text": [
- " threading : 3.7886 s\n",
- " processes : 5.2656 s\n",
- " sync : 14.7481 s\n"
- ]
- }
- ],
- "source": [
- "import time\n",
- "# -- dask testing different schedulers:\n",
- "for sch in ['threading', 'processes', 'sync']:\n",
- " t0 = time.time()\n",
- " r = yd.compute(scheduler=sch)\n",
- " t1 = time.time()\n",
- " print(f\"{sch:>10} : {t1 - t0:0.4f} s\")"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": 5,
- "id": "4c33d184-7c25-4052-aece-50dd9382e32e",
- "metadata": {},
- "outputs": [
- {
- "data": {
- "text/html": [
- "![\"High](\"https://docs.dask.org/en/stable/_images/dask-cluster-manager.svg\")
\n",
- "\n",
- "*Image credit: Dask Contributors*\n",
- "\n",
- "Dask Cluster Managers have different names corresponding to different computing environments. Some examples are `dask-jobqueue` for your HPC systems (including `PBSCluster`) or Kubernetes Cluster for machines on the Cloud. \n",
- "\n",
- "The NCAR tutorial series includes an in-depth exploration and practical use cases of Dask on HPC systems and best practices for Dask on HPC. For the complete set of NCAR tutorial materials, please refer to the main NCAR tutorial content available [here](https://ncar.github.io/dask-tutorial/README.html).\n",
- "\n",
- "For more information visit the [Dask Docs](https://docs.dask.org/en/stable/deploying.html).\n"
- ]
- }
- ],
- "metadata": {
- "kernelspec": {
- "display_name": "Python 3 (ipykernel)",
- "language": "python",
- "name": "python3"
- },
- "language_info": {
- "codemirror_mode": {
- "name": "ipython",
- "version": 3
- },
- "file_extension": ".py",
- "mimetype": "text/x-python",
- "name": "python",
- "nbconvert_exporter": "python",
- "pygments_lexer": "ipython3",
- "version": "3.9.18"
- },
- "widgets": {
- "application/vnd.jupyter.widget-state+json": {
- "state": {},
- "version_major": 2,
- "version_minor": 0
- }
- }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}
diff --git a/_preview/22/_sources/notebooks/how-to-cite.md b/_preview/22/_sources/notebooks/how-to-cite.md
deleted file mode 100644
index 7fce57a..0000000
--- a/_preview/22/_sources/notebooks/how-to-cite.md
+++ /dev/null
@@ -1,7 +0,0 @@
-# How to Cite This Cookbook
-
-The material in Project Pythia's Dask Cookbook is licensed for free and open consumption and reuse. All code is served under [Apache 2.0](https://www.apache.org/licenses/LICENSE-2.0), while all non-code content is licensed under [Creative Commons BY 4.0 (CC BY 4.0)](https://creativecommons.org/licenses/by/4.0/). Effectively, this means you are free to share and adapt this material so long as you give appropriate credit to the Cookbook authors and the Project Pythia community.
-
-The source code for the book is [released on GitHub](https://github.com/ProjectPythia/dask-cookbook) and archived on Zenodo. This DOI will always resolve to the latest release of the book source:
-
-[](https://zenodo.org/badge/latestdoi/610934658)
diff --git a/_preview/22/_sphinx_design_static/design-tabs.js b/_preview/22/_sphinx_design_static/design-tabs.js
deleted file mode 100644
index b25bd6a..0000000
--- a/_preview/22/_sphinx_design_static/design-tabs.js
+++ /dev/null
@@ -1,101 +0,0 @@
-// @ts-check
-
-// Extra JS capability for selected tabs to be synced
-// The selection is stored in local storage so that it persists across page loads.
-
-/**
- * @type {Record