Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

DOCS-#6949: Create Modin on Dask cluster tutorial #6950

Merged
merged 8 commits into from
Feb 27, 2024
Merged

DOCS-#6949: Create Modin on Dask cluster tutorial #6950

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
608cfb7
DOCS-#6949: Create Modin on Dask cluster tutorial
Retribution98 Feb 20, 2024
6391c34
Apply suggestions from code review
Retribution98 Feb 20, 2024
f5297e9
Fix after review
Retribution98 Feb 20, 2024
d690d5b
Apply suggestions from code review
Retribution98 Feb 26, 2024
d43b2e7
Apply suggestions from code review
Retribution98 Feb 26, 2024
8bcc1a7
fix suggestions
Retribution98 Feb 26, 2024
a27c527
add minor changes
Retribution98 Feb 26, 2024
9914b3b
Apply suggestions from code review
Retribution98 Feb 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
372 changes: 372 additions & 0 deletions examples/tutorial/jupyter/execution/pandas_on_dask/cluster/exercise_5.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,372 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"![LOGO](../../../img/MODIN_ver2_hrz.png)\n",
"\n",
"<h1>Scale your pandas workflows by changing one line of code</h2>"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Exercise 5: Setting up cluster environment\n",
"\n",
"**GOAL**: Learn how to set up a Dask cluster for Modin, connect Modin to a Dask cluster and run pandas queries on a cluster.\n",
"\n",
"**NOTE**: This exercise has extra requirements. Read instructions carefully before attempting. \n",
"\n",
"**This exercise instructs users on how to start a 500+ core Dask cluster, and it is not shut down until the end of exercise.**"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Often in practice we have a need to exceed the capabilities of a single machine. Modin works and performs well \n",
"in both local mode and in a cluster environment. The key advantage of Modin is that your python code does not \n",
"change between local development and cluster execution. Users are not required to think about how many workers \n",
"exist or how to distribute and partition their data; Modin handles all of this seamlessly and transparently.\n",
"\n",
"![Cluster](../../../img/modin_cluster.png)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Extra requirements for AWS authentication\n",
"\n",
"First of all, install the necessary dependencies in your environment:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"!pip install dask_cloudprovider[aws]"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"The next step is to setup your AWS credentials, namely, set ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``\n",
"and ``AWS_SESSION_TOKEN`` (Optional) (refer to [AWS CLI environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html) to get more insight on this):"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"\n",
"os.environ[\"AWS_ACCESS_KEY_ID\"] = \"<aws_access_key_id>\"\n",
"os.environ[\"AWS_SECRET_ACCESS_KEY\"] = \"<aws_secret_access_key>\"\n",
"os.environ[\"AWS_SESSION_TOKEN\"] = \"<aws_session_token>\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Starting and connecting to the cluster\n",
"\n",
"This example starts 1 scheduler node (m5.24xlarge) and 6 worker nodes (m5.24xlarge), 576 total CPUs. Keep in mind the scheduler node manages cluster operation but doesn't perform any execution.\n",
"\n",
"You can check the [Amazon EC2 pricing](https://aws.amazon.com/ec2/pricing/on-demand/) page.\n",
"\n",
"Dask cluster can be deployed in different ways (refer to [Dask documentaion](https://docs.dask.org/en/latest/deploying.html) to get more information about it), but in this tutorial we will use the ``EC2Cluster`` from [dask_cloudprovider](https://cloudprovider.dask.org/en/latest/) to create and initialize a Dask cluster on Amazon Web Service (AWS).\n",
"\n",
"**Note**: EC2Cluster uses a docker container to run the scheduler and each of the workers. Probably you need to use another docker image depending on your python version and requirements. You can find more docker-images on [daskdev](https://hub.docker.com/u/daskdev) page.\n",
"\n",
"In the next cell you can see how the EC2Cluster is being created. <b>Set your ``key_name`` and modify AWS settings as required before running it.</b>"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from dask_cloudprovider.aws import EC2Cluster\n",
"\n",
"n_workers = 6\n",
"cluster = EC2Cluster(\n",
" # AWS parameters\n",
" key_name = \"\", # set your keyname\n",
" region = \"us-west-2\",\n",
" availability_zone = [\"us-west-2a\"],\n",
" ami = \"ami-0387d929287ab193e\",\n",
" instance_type = \"m5.24xlarge\",\n",
" vpc = \"vpc-002bd14c63f227832\",\n",
" subnet_id = \"subnet-09860dafd79720938\",\n",
" filesystem_size = 200, # in GB\n",
"\n",
" # DASK parameters\n",
" n_workers = n_workers,\n",
" docker_image = \"daskdev/dask:latest\",\n",
" debug = True,\n",
" security=False,\n",
")\n",
"\n",
"scheduler_adress = cluster.scheduler_address\n",
"print(f\"Scheduler IP address of Dask cluster: {scheduler_adress}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"After creating the cluster you need to connect to it. To do this you should put the ``EC2Cluster`` instance or the scheduler IP address in ``distributed.Client``.\n",
"\n",
"When you connect to the cluster, the workers may not be initialized yet, so you need to wait for them using ``client.wait_for_workers``.\n",
"\n",
"Then you can call ``client.ncores()`` and check which workers are available and how many threads are used for each of them."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from distributed import Client\n",
"\n",
"client = Client(cluster)\n",
"# Or use an IP address connection if the cluster instance is unavailable:\n",
"# client = Client(f\"{scheduler_adress}:8687\")\n",
"\n",
"client.wait_for_workers(n_workers)\n",
"client.ncores()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"After successful initialization of the cluster, you need to configure it.\n",
"\n",
"You can use plugins to install any requirements into workers:\n",
"* [InstallPlugin](https://distributed.dask.org/en/stable/plugins.html#distributed.diagnostics.plugin.InstallPlugin)\n",
"* [PipInstall](https://distributed.dask.org/en/stable/plugins.html#distributed.diagnostics.plugin.PipInstall)\n",
"* [CondaInstall](https://distributed.dask.org/en/stable/plugins.html#distributed.diagnostics.plugin.CondaInstall).\n",
"\n",
"You have to install Modin package on each worker using ``PipInstall`` plugin."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from dask.distributed import PipInstall\n",
"\n",
"client.register_plugin(PipInstall(packages=[\"modin\"]))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"If you need an additional workers configuration, you can create your own [WorkerPlugin](https://distributed.dask.org/en/stable/plugins.html#worker-plugins) or function that will be executed on each worker upon calling ``client.run()``.\n",
"\n",
"**NOTE**: Dask cluster does not check if this plugin or function has been called before. Therefore, you need to take this into account when using them.\n",
"\n",
"In this tutorial a CSV file will be read, so you need to download it to each of the workers and local machine with the same global path."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from dask.distributed import Worker\n",
"import os\n",
"import urllib\n",
"\n",
"def dataset_upload(file_url, file_path):\n",
" try:\n",
" dir_name = os.path.dirname(file_path)\n",
" if not os.path.exists(dir_name):\n",
" os.makedirs(dir_name)\n",
" if os.path.exists(file_path):\n",
" return \"File has already existed.\"\n",
" else:\n",
" urllib.request.urlretrieve(file_url, file_path)\n",
" return \"OK\"\n",
" except Exception as ex:\n",
" return str(ex)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Set the directory where it should be downloaded (the local directory will be used by default):"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"directory_path = \"./\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Then you need to run `dataset_upload` function on all workers. As the result, you will get a dictionary, where the result of the function execution will be for each workers:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"file_path = os.path.join(os.path.abspath(directory_path), \"taxi.csv\")\n",
"client.run(dataset_upload, \"https://modin-datasets.intel.com/testing/yellow_tripdata_2015-01.csv\", file_path)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You have to also execute this function on the local machine:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"dataset_upload(\"https://modin-datasets.intel.com/testing/yellow_tripdata_2015-01.csv\", file_path)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>Congratulations! The cluster is now fully configured and we can start running Pandas queries.</b>"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Executing in a cluster environment\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Same as local mode Modin on cluster uses Ray as an execution engine by default so no additional action is required to start to use it. Alternatively, if you need to use another engine, it should be specified either by setting the Modin config or by setting Modin environment variable before the first operation with Modin as it is shown below. Also, note that the full list of Modin configs and corresponding environment variables can be found in the [Modin Configuration Settings](https://modin.readthedocs.io/en/stable/flow/modin/config.html#modin-configs-list) section of the Modin documentation."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Modin engine can be specified either by config\n",
"import modin.config as cfg\n",
"cfg.Engine.put(\"dask\")\n",
"\n",
"# or by setting the environment variable\n",
"# import os\n",
"# os.environ[\"MODIN_ENGINE\"] = \"dask\""
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Now you can use Modin on the Dask cluster.\n",
"\n",
"Let's read the downloaded CSV file and execute such pandas operations as count, groupby and map:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import modin.pandas as pd\n",
"import time\n",
"\n",
"t0 = time.perf_counter()\n",
"\n",
"df = pd.read_csv(file_path, quoting=3)\n",
"df_count = df.count()\n",
"df_groupby_count = df.groupby(\"passenger_count\").count()\n",
"df_map = df.map(str)\n",
"\n",
"t1 = time.perf_counter()\n",
"print(f\"Full script time is {(t1 - t0):.3f}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Shutting down the cluster\n",
"\n",
"Now that we have finished computation, we can shut down the cluster:"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"cluster.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### This ends the cluster exercise"
]
}
],
"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.12"
}
},
"nbformat": 4,
"nbformat_minor": 2
}