thêm chức năng train trên odc predict trên planetary
This commit is contained in:
Victor Phan
@@ -0,0 +1,696 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "416ee5f1-2a79-4952-a901-0d1f027f468d",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"# Introduction to Dask and the Open Data Cube <img align=\"right\" src=\"../../resources/csiro_easi_logo.png\">\n",
|
||||
"\n",
|
||||
"**Prerequisites**: This material assumes basic knowledge of the Open Data Cube, Xarray and numerical processing using numpy.\n",
|
||||
"\n",
|
||||
"- [Introduction](#Introduction-to-Dask-and-the-Open-Data-Cube)\n",
|
||||
" - [Planning and writing efficient applications](#Planning-and-writing-efficient-applications)\n",
|
||||
" - [What you will learn](#What-you-will-learn)\n",
|
||||
"- [Performance in Python](#Performance-in-Python)\n",
|
||||
" - [Python list and numpy](#Python-list-and-numpy)\n",
|
||||
" - [Numba - accelerating Python](#Numba---accelerating-Python)\n",
|
||||
"- [Parallelism with Dask](#Parallelism-with-Dask)\n",
|
||||
" - [Review of dask](#Review-of-dask)\n",
|
||||
" - [Dask local cluster](#Dask-local-cluster)\n",
|
||||
"\n",
|
||||
"The Open Data Cube library is written in Python and makes extensive use of scientific and geospatial libraries. For the purposes of this tutorial we will primarily consider five libraries:\n",
|
||||
"\n",
|
||||
" 1. `datacube` - EO datacube\n",
|
||||
" 1. `xarray` - labelled arrays\n",
|
||||
" 1. (optional) `dask` & `distributed` - distributed parallel programming\n",
|
||||
" 1. `numpy` - numerical array processing with vectorisation\n",
|
||||
" 1. (optional) `numba` - a library for high performance python\n",
|
||||
"\n",
|
||||
"Whilst the interrelations are intimate it is useful to conceptualise them according to their primary role and how these roles build from low level numerical array processing (`numpy`) through to high-level EO datacube semantics (`datacube` and `xarray`). If you prefer, viewed from top to bottom we can say:\n",
|
||||
" 1. `datacube.load()` does the necessary file IO and data manipulation to construct a...\n",
|
||||
" 1. `xarray` which will be labelled with the necessary coordinate systems and band names and made up of...\n",
|
||||
" 1. (optionally) `dask.array`s which contain many `chunks` which are...\n",
|
||||
" 1. `numpy` arrays containing the actual data values.\n",
|
||||
"\n",
|
||||
"Each higher level of abstraction thus builds on the lower level components that perform the actual storage and computation.\n",
|
||||
"\n",
|
||||
"Overlaid on this are libraries like `numba`, `dask` and `distributed` that provide computational components that can accelerate and distribute processing across multiple compute cores and computers. The use of `dask`, `distributed` and `numba` are optional - not all applications require the additional complexity of these tools.\n",
|
||||
"\n",
|
||||
"### Planning and writing efficient applications\n",
|
||||
"\n",
|
||||
"Achieving performance and scale requires an understanding of the performance of each library and how it interacts with the others. Moreover, and often counterintuitively, adding more compute cores to a problem may not make it faster; in fact it may slow down (as well as waste resources). Added to that is the _deceptive simplicity_ in that some of the tools can be simply _turned on_ with only a few code changes and significant performance increases can be achieved.\n",
|
||||
"\n",
|
||||
"However, as the application is scaled or an alternative algorithm is used further challenges may arise, in expected I/O or compute efficiency, that require code refactors and changes in algorithmic approach. These challenges can seem to undo some of the earlier work and be frustrating to address. \n",
|
||||
"\n",
|
||||
"The good news is whilst there is complexity (six interrelated libraries mentioned so far), there are common concepts and techniques involved in analysing _how to optimise your algorithm_. If you know from the start your application is going to require scale, then it does help to think in advance where you are heading.\n",
|
||||
"\n",
|
||||
"### What you will learn\n",
|
||||
"\n",
|
||||
"This course will equip readers with concepts and techniques they can utilise in their algorithm and workflow development. The course will be using computer science terms and a variety of libraries but won't be discussing these in detail in order to keep this course concise. The focus will be on demonstration by example and analysis techniques to identify where to focus effort. The reader is encouraged to use their favorite search engine to dig deeper when needed; there are a lot of tutorials online!\n",
|
||||
"\n",
|
||||
"One last thing, in order to maintain a healthy state of mind for \"Dask and ODC\", the reader is encouraged to hold both of these truths in mind at the same time:\n",
|
||||
" 1. The *best* thing about dask is it makes distributed parallel programming in the datacube easy\n",
|
||||
" 1. The *worst* thing about dask it is makes distributed parallel programming in the datacube easy\n",
|
||||
"\n",
|
||||
"Yep, that's contradictory! By the end of this course, and a couple of your own adventures, you will understand why."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "f3fb9221-d423-4a3e-baa4-6cdaefd56c1d",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# EASI defaults\n",
|
||||
"import git\n",
|
||||
"import sys\n",
|
||||
"import os\n",
|
||||
"os.environ['USE_PYGEOS'] = '0'\n",
|
||||
"repo = git.Repo('.', search_parent_directories=True).working_tree_dir\n",
|
||||
"if repo not in sys.path: sys.path.append(repo)\n",
|
||||
"from easi_tools import EasiDefaults, notebook_utils"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "78b8349c-6698-45d2-a159-8013222232d1",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"easi = EasiDefaults()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "5be4a71f-d1f0-4c2a-a0cb-f53757d78cb1",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"# Performance in Python\n",
|
||||
"\n",
|
||||
"In this section we will explore python performance for array processing. Python itself, as you will soon see, is quite slow. It is, however, highly expressive and can orchestrate more complex and faster libraries of numerical code (e.g., `numpy`). Python is also ammendable to being accelerated (e.g. using `numba`) and made to run on multiple CPU cores (e.g. via `dask`). \n",
|
||||
"\n",
|
||||
"### Python `list` and `numpy`\n",
|
||||
"\n",
|
||||
"Let's take a look at the simple addition of two arrays. In Python the nearest data type to an array is a `list` of numbers. This will be our starting point.\n",
|
||||
"\n",
|
||||
"Our focus is on performance so we'll use the Jupyter `%%time` and `%%timeit` magics to run our cells and time their execution. The latter will run the cell multiple times and provide us with more representative statistics of performance and variability.\n",
|
||||
"\n",
|
||||
"First in pure Python using lists :"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "f2dfa3fd-0cc7-4804-8790-ed65f05eaa4e",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"size_of_vec = 2000*2000\n",
|
||||
"X_list = range(size_of_vec)\n",
|
||||
"Y_list = range(size_of_vec)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "d8c79652-2a47-40f4-87a2-8e90ce9cdf9c",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%timeit -n 10 -r 1\n",
|
||||
"Z = [X_list[i] + Y_list[i] for i in range(len(X_list)) ]"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "9aa9586d-b6ab-49d6-8914-ff016023cae8",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Now the same processing using `numpy`."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "5158c9d0-d8a9-48e6-ab1f-df4948dde092",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import numpy\n",
|
||||
"X = numpy.arange(size_of_vec)\n",
|
||||
"Y = numpy.arange(size_of_vec)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "a4dc338b-d411-40be-8aa0-4b50bce2201a",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%timeit -n 10 -r 1\n",
|
||||
"Z = X + Y"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "c8e8fb26-0d58-4dc1-b8ea-6c5ad044e4ea",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Let's check that the two arrays are identical (note that %%timeit does not make the variables above available due to the way that %%timeit works, so we reconstruct the arrays)."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "476b9020-ba06-47d4-84f7-f6ede912ce88",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"print(\"Are the arrays identical?\")\n",
|
||||
"([X_list[i] + Y_list[i] for i in range(len(X_list)) ] == X + Y).all()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "7ff1825c-9bf8-4ab8-baa9-13b43f4f4a3f",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"At least two orders of magnitude in performance improvement!\n",
|
||||
"\n",
|
||||
"Why?\n",
|
||||
"\n",
|
||||
"`numpy` provides a python interface to an underlying C array library that makes use of CPU `vectorization` - this allows it to process several add operations at the same time.\n",
|
||||
"\n",
|
||||
"`numpy` isn't the only library that does this type of wrapping over a fast optimised library. There are, for example,\n",
|
||||
"- `cuPy` which uses GPUs for array processing\n",
|
||||
"- `tensorflow` uses both CPU and GPU optimisations for machine learning\n",
|
||||
"- `datashader` for large dataset visualisation\n",
|
||||
"\n",
|
||||
"It's a very long list and thanks to a great deal of work by a great many software engineers most of these libraries will work together efficiently. \n",
|
||||
"\n",
|
||||
"> __Tip__: Where possible use high performance libraries that have python wrappers.\n",
|
||||
"\n",
|
||||
"The reader will have noticed the change in abstraction. The pure Python version used list comprehension syntax to add the two arrays, while `numpy` was a much shorter direct addition syntax more in keeping with the mathematics involved. This change in abstraction is seen in most libraries, including the ODC library where `datacube.load()` is shorthand for a complex process of data discovery, reprojection, fusing and array construction. High-level abstractions like this are powerful and greatly simplify development (the good). They can also hide performance bottlenecks and challenges (the bad).\n",
|
||||
"\n",
|
||||
"> __Tip__: Use high level API abstractions but be mindful of their use."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "2e01d98d-0090-46a7-adee-52534ee49f74",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"### `Numba` - accelerating Python\n",
|
||||
"\n",
|
||||
"So high performance libraries rock, but what if you don't have one for your purpose and you're back in Python?\n",
|
||||
"`Numba` translates Python functions into optimized machine code at runtime - https://numba.pydata.org.\n",
|
||||
"\n",
|
||||
"Let's see how this works. A more complex example this time with a smoothing function applied over our (random) image, perform an FFT, and save the result.\n",
|
||||
"These examples are (very) slightly modified versions from the [High Performance Python Processing Pipeline video by Matthew Rocklin](https://youtu.be/wANQkgDuTAk). It's such a good introduction its worth repeating.\n",
|
||||
"\n",
|
||||
"We'll also use the `tqdm` library to provide a progress bar."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "30019636-0446-4f0a-85ec-d93139065f74",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import numpy as np\n",
|
||||
"from tqdm.notebook import tqdm\n",
|
||||
"\n",
|
||||
"def load_eo_data():\n",
|
||||
" return np.random.random((1000, 1000))\n",
|
||||
"\n",
|
||||
"def smooth(x):\n",
|
||||
" out = np.empty_like(x)\n",
|
||||
" for i in range(1, x.shape[0] - 1):\n",
|
||||
" for j in range(1, x.shape[1] - 1):\n",
|
||||
" out[i, j] = (x[i + -1, j + -1] + x[i + -1, j + 0] + x[i + -1, j + 1] +\n",
|
||||
" x[i + 0, j + -1] + x[i + 0, j + 0] + x[i + 0, j + 1] +\n",
|
||||
" x[i + 1, j + -1] + x[i + 1, j + 0] + x[i + 1, j + 1]) // 9\n",
|
||||
" return out\n",
|
||||
"\n",
|
||||
"def save(x, filename):\n",
|
||||
" pass "
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "f83fbe3b-5138-472e-a6f2-7e34f71804bb",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%time\n",
|
||||
"for i in tqdm(range(5)):\n",
|
||||
" img = load_eo_data()\n",
|
||||
" img = smooth(img)\n",
|
||||
" img = np.fft.fft2(img)\n",
|
||||
" save(img, \"file-\" + str(i) + \"-.dat\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "6cc2f3e8-c9e5-4cf2-bb66-bb27c3a7211b",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"The `smooth(x)` function contains two python loops. Now we could (and would) find a similar high performance library with a `smooth(x)` function but for this example let's use `numba`'s `jit` compiler to translate the python function into optimized machine code at runtime."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "18808bb2-ca62-425c-9e08-f876ceeaaff2",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import numba\n",
|
||||
"\n",
|
||||
"fast_smooth = numba.jit(smooth)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "14d8a5e0-30f9-4676-a436-e5643c92dddf",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%time\n",
|
||||
"\n",
|
||||
"for i in tqdm(range(5)):\n",
|
||||
" img = load_eo_data()\n",
|
||||
" img = fast_smooth(img)\n",
|
||||
" img = np.fft.fft2(img)\n",
|
||||
" save(img, \"file-\" + str(i) + \"-.dat\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "82412220-68b4-4905-b6f0-2b53b71c55f5",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Just a bit quicker! Much of the time in the first run was `numba` performing compilation. Run the cell above again and you'll find it runs faster the second time.\n",
|
||||
"\n",
|
||||
"The _recommended_ approach to have `numba` compile a python function is to use python decorator syntax (`@numba.jit`). So the original code now looks like this (single line changed) and we can call `smooth(x)` without having to create `fast_smooth`:"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "820dd545-bae8-477b-9e1e-73c806b953bd",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import numpy as np\n",
|
||||
"\n",
|
||||
"def load_eo_data():\n",
|
||||
" return np.random.random((1000, 1000))\n",
|
||||
"\n",
|
||||
"@numba.jit\n",
|
||||
"def smooth(x):\n",
|
||||
" out = np.empty_like(x)\n",
|
||||
" for i in range(1, x.shape[0] - 1):\n",
|
||||
" for j in range(1, x.shape[1] - 1):\n",
|
||||
" out[i, j] = (x[i + -1, j + -1] + x[i + -1, j + 0] + x[i + -1, j + 1] +\n",
|
||||
" x[i + 0, j + -1] + x[i + 0, j + 0] + x[i + 0, j + 1] +\n",
|
||||
" x[i + 1, j + -1] + x[i + 1, j + 0] + x[i + 1, j + 1]) // 9\n",
|
||||
" return out\n",
|
||||
"\n",
|
||||
"def save(x, filename):\n",
|
||||
" pass"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "59f9604e-e0d6-4ac0-bd54-b93f6952131e",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%time\n",
|
||||
"for i in tqdm(range(5)):\n",
|
||||
" img = load_eo_data()\n",
|
||||
" img = smooth(img)\n",
|
||||
" img = np.fft.fft2(img)\n",
|
||||
" save(img, \"file-\" + str(i) + \"-.dat\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "4598e9d1-e154-4309-9bc0-0d5252775975",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Why not use `numba` all the time everywhere?\n",
|
||||
"\n",
|
||||
"Like most high level abstractions `numba` makes assumption about code, only accelerates a subset of python libraries (not all `numpy` functions are available via `numba`), and it is entirely possible it can make performance worse or not work at all!\n",
|
||||
"\n",
|
||||
"There's one additional consideration. If you've run all the cells to this point in order, try running the `fast_smooth` cell again, repeated below for convenience:\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "01a91155-67a3-4ab2-a2ff-712d0c43c71e",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"fast_smooth = numba.jit(smooth)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "9675cb05-6794-4e9f-8daa-099846dfd3d0",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"source": [
|
||||
"Error!\n",
|
||||
"\n",
|
||||
"The `smooth` function was decorated so is already `jit`-compiled. Attempting to do so again causes this error, and exposes some of the low level changes behind the abstraction.\n",
|
||||
"This can make debugging code difficult if you are not mindful of what is occuring.\n",
|
||||
"\n",
|
||||
"TIP: __Use high level API abstractions but be mindful of their use__"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "25419e83-3fce-49be-b1b4-54f6415d7096",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"# Parallelism with Dask\n",
|
||||
"\n",
|
||||
"Our fake EO processing pipeline only has 5 images and takes about 1 sec to run. In practice we'll have 1000s of images to process (if not more).\n",
|
||||
"\n",
|
||||
"Let's repeat our example code but now with more iterations. You can understand why we use The `tqdm` library to provide a progress bar for these larger scale examples rather than printing out each iteration number or staring at a blank screen wondering if it works!\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "af003cf6-3993-4718-a0d0-48c34f96c0e4",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import numpy as np\n",
|
||||
"import numba\n",
|
||||
"from tqdm.notebook import tqdm\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"def load_eo_data():\n",
|
||||
" return np.random.random((1000, 1000))\n",
|
||||
"\n",
|
||||
"@numba.jit\n",
|
||||
"def smooth(x):\n",
|
||||
" out = np.empty_like(x)\n",
|
||||
" for i in range(1, x.shape[0] - 1):\n",
|
||||
" for j in range(1, x.shape[1] - 1):\n",
|
||||
" out[i, j] = (x[i + -1, j + -1] + x[i + -1, j + 0] + x[i + -1, j + 1] +\n",
|
||||
" x[i + 0, j + -1] + x[i + 0, j + 0] + x[i + 0, j + 1] +\n",
|
||||
" x[i + 1, j + -1] + x[i + 1, j + 0] + x[i + 1, j + 1]) // 9\n",
|
||||
" return out\n",
|
||||
"\n",
|
||||
"def save(x, filename):\n",
|
||||
" pass"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "a44bb7e4-7f70-40a1-84cc-9b2f63007378",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Before running the next code, open a terminal window (File>New>Terminal) and run `htop` at the command line to show current CPU usage per core.\n",
|
||||
"\n",
|
||||
"> **TIP:** Drag your terminal window so that it sits below this notebook before you run `htop` to see both windows at the same time."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "88d0fc98-eb4a-445e-b305-84aa493d07b6",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%time\n",
|
||||
"for i in tqdm(range(1000)):\n",
|
||||
" img = load_eo_data()\n",
|
||||
" img = smooth(img)\n",
|
||||
" img = np.fft.fft2(img)\n",
|
||||
" save(img, \"file-\" + str(i) + \"-.dat\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "c2ef6e05-6466-4119-b33d-eaa64445ba24",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"You'll notice that only one core is showing any load. The above code is not using any of the additional cores.\n",
|
||||
"\n",
|
||||
"`dask` can be useful in this scenario even on a local machine. "
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "d7fcafa1-75cf-4856-987d-bf1154203a01",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"### Review of dask\n",
|
||||
"\n",
|
||||
"Firstly, a few notes on terminology. A Dask Cluster is comprised of a __client__, a __scheduler__, and __workers__. These terms will be used throughout this tutorial. Figure 1 below shows the relationship between each of these components. The __client__ submits tasks to the __scheduler__, which decides how to submit the tasks to individual workers. During this process, the scheduler creates what is called a __Task Graph__. This is essentially a map of the tasks that need to be carried out. Figure 2 shows an example of a simple task graph (see https://docs.dask.org/en/stable/graphs.html for more information). __Workers__ carry out the actual calculations and either store the results or send them back to the client.\n",
|
||||
"\n",
|
||||
"<div>\n",
|
||||
" <span style=\"border:solid 1px #888;float:left;padding:10px;margin-right:25px;width:550px\">\n",
|
||||
" <img src=\"../../resources/distributed-overview.png\">\n",
|
||||
" <figcaption><em>Figure 1. Overview of a Dask Cluster.</em></figcaption>\n",
|
||||
" </span>\n",
|
||||
" <span style=\"border:solid 1px #888;float:left;padding:10px;margin-left:25px;width:150px\">\n",
|
||||
" <img style=\"float:left\" src=\"../../resources/dask-simple.png\">\n",
|
||||
" <figcaption><em>Figure 2. A simple Task Graph.</em></figcaption>\n",
|
||||
" </span>\n",
|
||||
"</div>"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "5d4cae0a-7557-440e-b6b0-54ffd968cced",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Dask has several core data types, including __Dask DataFrames__ and __Dask Arrays__. Essentially, Dask DataFrames are parallelized Pandas DataFrames (Figure 3) and Dask Arrays are parallelized Numpy arrays (Figure 4).\n",
|
||||
"\n",
|
||||
"<div style=\"width:100%\">\n",
|
||||
" <span style=\"border:solid 1px #888;float:left;padding:10px;margin-right:25px;width:200px\">\n",
|
||||
" <img src=\"../../resources/dask-dataframe.svg\">\n",
|
||||
" <figcaption><em>Figure 3. A Dask DataFrame is comprised of many in-memory pandas DataFrames separated along an index.</em></figcaption>\n",
|
||||
" </span>\n",
|
||||
" <span style=\"border:solid 1px #888;float:left;padding:10px;margin-left:25px;width:300px\">\n",
|
||||
" <img style=\"float:left\" src=\"../../resources/dask-array.svg\">\n",
|
||||
" <figcaption><em>Figure 4. A Dask Array is a subset of the NumPy <code>ndarray</code> interface using blocked algorithms, cutting up the large array into many small arrays.</em></figcaption>\n",
|
||||
" </span>\n",
|
||||
"</div>"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "d3f64100-b5cc-48a8-9be9-71f31e50dd35",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"EASI and the Open Data Cube primarily make use of __Dask DataArrays__. For more information see https://tutorial.dask.org/02_array.html."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "823a9c12-1c03-4ea8-8a9e-ec498bdbc056",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"### Dask local cluster\n",
|
||||
"\n",
|
||||
"Let's start by creating a local dask cluster."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "e0d95f91-a9e0-44c1-a577-f8e708ccbce6",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"from dask.distributed import Client, LocalCluster, fire_and_forget\n",
|
||||
"\n",
|
||||
"cluster = LocalCluster()\n",
|
||||
"client = Client(cluster)\n",
|
||||
"client"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "e96378b8-ec25-4970-9957-8c482112073a",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"There is also a utility function in `notebook_utils` which you can use to initialize a local dask cluster. This funtion can also be used to initialize a remote cluster using Dask Gateway, but please complete the other Dask tutorials before using a remote cluster.\n",
|
||||
"\n",
|
||||
"The following lines will initialize and display a local dask cluster and can replace the code above.\n",
|
||||
"\n",
|
||||
"```python\n",
|
||||
"cluster, client = notebook_utils.initialize_dask(use_gateway=False, wait=True)\n",
|
||||
"display(cluster if cluster else client)\n",
|
||||
"```"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "922c7673-88cf-445c-90c7-ea4c6c2124d9",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"The Dask Dashboard url will show as \"localhost\" or \"127.0.0.1\" since its running locally in the Jupyter kernel. The dashboard can be accessed via the Jupyter server proxy using the following url:"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "31a07d9e-3b83-4081-9db2-cd70ac3853d9",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"notebook_utils.localcluster_dashboard(client=client, server=easi.hub)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "bd3c526a-c04f-40ac-b2dc-46701f9ec739",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"You will want to have this open when running the next cell.\n",
|
||||
"\n",
|
||||
"We modify our application to `submit` functions to run on the dask cluster:"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "c8b09db6-aea0-4877-a0e8-4cc2497c8936",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"%%time\n",
|
||||
"\n",
|
||||
"for i in tqdm(range(1000)):\n",
|
||||
" img = client.submit(load_eo_data, pure=False)\n",
|
||||
" img = client.submit(smooth, img)\n",
|
||||
" img = client.submit(np.fft.fft2, img)\n",
|
||||
" future = client.submit(save, img, \"file-\" + str(i) + \"-.dat\")\n",
|
||||
" fire_and_forget(future)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "032d7d77-3e37-4fb5-bd9d-fa7f1c6db1ff",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"If you watch `htop` in the terminal you'll see all cores become active. The dask dashboard will also provide a view of the tasks being run in parallel."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "ef6c1656-f68b-4192-a88d-21a8d0b78c7e",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"You will also see that the cluster remains busy after the cell above finishes. This is because Dask is working in the background processing in parallel the tasks that have been submitted to it."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"id": "bd40a084-1721-4d10-9400-e2bd563f5823",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"A `dask.distributed.LocalCluster()` will shutdown when this notebook kernel is stopped.\n",
|
||||
"Still it's a good practice to close the client and the cluster so its all cleaned up. This will be more important when using dask distributed clusters as they are independent of the notebook kernel."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "70edbc54-df60-4f33-8b56-9a10a8bb3fa7",
|
||||
"metadata": {
|
||||
"tags": []
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"client.close()\n",
|
||||
"\n",
|
||||
"cluster.close()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"id": "8c6e908d-1467-4246-bf12-e1a2cfd07be2",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": []
|
||||
}
|
||||
],
|
||||
"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.10.6"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 5
|
||||
}
|
||||
Reference in New Issue
Block a user