Ray_Core/Ray_Core_1_Remote_Functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a14cab54",
   "metadata": {},
   "source": [
    "# A Guided Tour of Ray Core: Remote Tasks\n",
    "\n",
    "© 2019-2023, Anyscale. All Rights Reserved\n",
    "\n",
    "## Introduction\n",
    "\n",
    "Ray enables arbitrary Python functions to be executed asynchronously on separate Python workers. These asynchronous Ray functions are called “tasks.” You can specify task's resource requirements in terms of CPUs, GPUs, and custom resources. These resource requests are used by the cluster scheduler to distribute tasks across the cluster for parallelized execution.  \n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/python_to_ray_concept_map.png\" height=\"55%\" width=\"50%\">|\n",
    "|:--|\n",
    "|Transforming Python code into Ray Tasks, Actors, and Immutable Ray objects.|\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/python_to_ray_task_map.png\" height=\"55%\" width=\"50%\">|\n",
    "|:--|\n",
    "|Transforming Python function into Ray Tasks|\n",
    "\n",
    "## Learning objectives\n",
    "In this this tutorial, you'll learn about:\n",
    " * Remote Task Parallel Pattern\n",
    " * Stateless remote functions as distributed tasks\n",
    " * Serial vs Parallel execution \n",
    " * Understand the concept of a Ray task \n",
    " * Easy API to convert an existing Python function into a Ray remote task\n",
    " * Walk through examples comparing serial vs. distributed Python functions and Ray tasks respectively"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bee29917",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import logging\n",
    "import math\n",
    "import random\n",
    "\n",
    "from pathlib import Path\n",
    "from typing import Tuple, List\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import pyarrow.parquet as pq\n",
    "import tqdm\n",
    "import ray"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "74b98b39",
   "metadata": {
    "tags": []
   },
   "source": [
    "## 1. Tasks Parallel Pattern\n",
    "\n",
    "Ray converts decorated functions with `@ray.remote` into stateless tasks, scheduled anywhere on a Ray node's worker in the cluster. \n",
    "\n",
    "Where they will be executed on the cluster (and on what node by which worker process), you don't have to worry about its details. All that is taken care for you. Nor do \n",
    "you have to reason about it — all that burden is Ray's job. You simply take your existing Python functions and covert them into \n",
    "distributed stateless *Ray Tasks*: **as simple as that!**"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "893ec22c",
   "metadata": {},
   "source": [
    "### Serial vs Parallelism Execution\n",
    "\n",
    "Serial tasks as regular Python functions are executed in a sequential manner, as shown\n",
    "in the diagram below. If I launch ten tasks, they will run on a single worker, one after the other.\n",
    " \n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Overview_of_Ray/sequential_timeline.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Timeline of sequential tasks, one after the other.|\n",
    "\n",
    "Compared to serial execution, a Ray task executes in parallel, scheduled on different workers. The Raylet will schedule these task based on [scheduling policies.](https://docs.ray.io/en/latest/ray-core/scheduling/index.html#ray-scheduling-strategies)\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Overview_of_Ray/distributed_timeline.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Sample timeline with ten tasks running across 4 worker nodes in parallel.|\n",
    "\n",
    "Let's look at some tasks running serially and then in parallel. For illustration, we'll use a the following tasks:\n",
    " * Generating fibonacci numbers serially and distributed\n",
    " * Computing value of pi using the monte carlo method\n",
    " * Transforming and processing large high-resolution images\n",
    " * Doing batch inference using Ray tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93a44dfe",
   "metadata": {},
   "source": [
    "But first, some basic concepts: There are a few key differences between an original Python function and the decorated one:\n",
    "\n",
    "**Invocation**: The regular version is called with `func_name()`, whereas the remote Ray version is called with `func_name.remote()`. Keep this pattern in mind for all Ray remote execution methods.\n",
    "\n",
    "**Mode of execution and return values**: A Python `func_name()` executes synchronously and returns the result of the function, whereas a Ray task `func_name.remote()` immediately returns an `ObjectRef` (a future) and then executes the task in the background on a remote worker process. \n",
    "\n",
    "The result of the future is obtained by calling `ray.get(ObjectRef)` on the `ObjectRef`. This is a blocking function."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b04aec4",
   "metadata": {},
   "source": [
    "Let's launch a Ray cluster on our local machine."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4aad47d9",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized:\n",
    "    ray.shutdown()\n",
    "ray.init(logging_level=logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "27eeba1a",
   "metadata": {},
   "source": [
    "### Example 1: Generating Fibonnaci series\n",
    "\n",
    "Let's define two functions: one runs serially, the other runs on a Ray cluster (local or remote). This example is borrowed and refactored from our \n",
    "blog: [Writing your First Distributed Python Application with Ray](https://www.anyscale.com/blog/writing-your-first-distributed-python-application-with-ray). \n",
    "(This is an excellent tutorial to get started with the concept of why and when to use Ray tasks and Ray Actors. Highly recommended read!)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "53bb49fa-00c0-4994-bd74-2d1093ff5d3b",
   "metadata": {},
   "outputs": [],
   "source": [
    "SEQUENCE_SIZE = 100000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "49994253",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function for local execution \n",
    "def generate_fibonacci(sequence_size):\n",
    "    fibonacci = []\n",
    "    for i in range(0, sequence_size):\n",
    "        if i < 2:\n",
    "            fibonacci.append(i)\n",
    "            continue\n",
    "        fibonacci.append(fibonacci[i-1]+fibonacci[i-2])\n",
    "    return len(fibonacci)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92777c6f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function for remote Ray task with just a wrapper\n",
    "@ray.remote\n",
    "def generate_fibonacci_distributed(sequence_size):\n",
    "    return generate_fibonacci(sequence_size)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b8b6ff1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the number of cores \n",
    "os.cpu_count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "675f4fb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Normal Python in a single process \n",
    "def run_local(sequence_size):\n",
    "    results = [generate_fibonacci(sequence_size) for _ in range(os.cpu_count())]\n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "45a917b1-7af2-4729-8b93-39883fc5054e",
   "metadata": {},
   "source": [
    "### Run in serial mode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "199e8e69",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "run_local(SEQUENCE_SIZE)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3fc510e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Distributed on a Ray cluster\n",
    "def run_remote(sequence_size):\n",
    "    results = ray.get([generate_fibonacci_distributed.remote(sequence_size) for _ in range(os.cpu_count())])\n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8fda02ce-cafd-4183-b36c-6aa28234cdf4",
   "metadata": {},
   "source": [
    "### Run as distributed Ray tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b4ebb951",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "run_remote(SEQUENCE_SIZE)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7bcecd59-8b2c-4983-acc1-9572ba04590e",
   "metadata": {},
   "source": [
    "### Recap\n",
    "As you can see that running as Ray Tasks, we see a significant performance improvment\n",
    "📈 by simply adding a Python decorator `ray.remote(...)`.\n",
    "\n",
    "To see how different values of computing Fibonnacci number affects the serial vs. performance execution times, try the exercise below."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "82484218-15bb-4e4f-a538-c8a75ee2ee71",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "1. Increase the fibonacci with 200K, 300K and observe the execution times"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ec375bb-d463-496d-90ec-0f8aa64d597f",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f12e65f5-5df7-4155-89cd-89e836d6c778",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "SEQUENCE_SIZE = 200000, 300000"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9ea4b7d2-2ceb-4a87-9a7b-3945f15f7d0a",
   "metadata": {},
   "source": [
    "### Example 2:  Monte Carlo simulation of estimating π\n",
    "\n",
    "Let's estimate the value of π using a [Monte Carlo](https://en.wikipedia.org/wiki/Monte_Carlo_method) method. We randomly sample points within a 2x2 square. We can use the proportion of the points that are contained within the unit circle centered at the origin to estimate the ratio of the area of the circle to the area of the square. \n",
    "\n",
    "Given we know that the true ratio to be π/4, we can multiply our estimated ratio by 4 to approximate the value of π. The more points that we sample to calculate this approximation, the closer we get to true value of π to required decimal points.\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/monte_carlo_pi.png\" width=\"80%\" height=\"80%\">|\n",
    "|:--|\n",
    "|Estimating the value of π by sampling random points that fall into the circle.|"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "373b1b0b-7ecd-4642-96e3-bcd8724172a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Change this to match your cluster scale.\n",
    "NUM_SAMPLING_TASKS = os.cpu_count()\n",
    "NUM_SAMPLES_PER_TASK = 10_000_000\n",
    "TOTAL_NUM_SAMPLES = NUM_SAMPLING_TASKS * NUM_SAMPLES_PER_TASK"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba80a46c-af65-4a0c-8078-371ef9e17c72",
   "metadata": {},
   "source": [
    "Define a regular function that computes the number of samples\n",
    "in the circle. This is done by randomly sampling `num_samples` for\n",
    "x, y between a uniform value of (-1, 1). Using the [math.hypot](https://docs.python.org/3/library/math.html#math.hypot) function, we\n",
    "compute if it falls within the circle."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c9846144-36cd-43d6-a465-1bfd686398bf",
   "metadata": {},
   "outputs": [],
   "source": [
    "def sampling_task(num_samples: int, task_id: int, verbose=True) -> int:\n",
    "    num_inside = 0\n",
    "    for i in range(num_samples):\n",
    "        x, y = random.uniform(-1, 1), random.uniform(-1, 1)\n",
    "        # check if the point is inside the circle\n",
    "        if math.hypot(x, y) <= 1:\n",
    "            num_inside += 1\n",
    "    if verbose:\n",
    "        print(f\"Task id: {task_id} | Samples in the circle: {num_inside}\")\n",
    "    return num_inside"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85c5b9d5-2aa7-4abe-aea5-15aa748cd055",
   "metadata": {},
   "source": [
    "Define a function to run this serially, by launcing `NUM_SAMPLING_TASKS` serial tasks in a comprehension list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ca8fe02-a7a4-4522-a2dd-cc57c471bce6",
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_serial(sample_size) -> List[int]:\n",
    "    results = [sampling_task(sample_size, i+1) for i in range(NUM_SAMPLING_TASKS)]\n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d195094-5c50-40ce-8640-38749acb7d18",
   "metadata": {},
   "source": [
    "Define a function to run this as a remote Ray task, which invokes our sampling function, but since it's decorated\n",
    "with `@ray.remote`, the task will run on a worker process, tied to a core, on the Ray cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "964a3647-318d-4395-bedc-fa1d825cda4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def sample_task_distribute(sample_size, i) -> object:\n",
    "    return sampling_task(sample_size, i)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0369350-0b0a-4486-9710-db89dccddf31",
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_disributed(sample_size) -> List[int]:\n",
    "    # Launch Ray remote tasks in a comprehension list, each returns immediately with a future ObjectRef \n",
    "    # Use ray.get to fetch the computed value; this will block until the ObjectRef is resolved or its value is materialized.\n",
    "    results = ray.get([\n",
    "            sample_task_distribute.remote(sample_size, i+1) for i in range(NUM_SAMPLING_TASKS)\n",
    "        ])\n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b0c52ce-348f-4239-a9e3-e9802e56e6ad",
   "metadata": {},
   "source": [
    "Define a function to calculate the value of π by getting all number of samples inside the circle from the sampling tasks and calculate π."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b94e50ae-0a0f-4c9e-9d8c-9220b04560d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_pi(results: List[int]) -> float:\n",
    "    total_num_inside = sum(results)\n",
    "    pi = (total_num_inside * 4) / TOTAL_NUM_SAMPLES\n",
    "    return pi"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aa9969a3-8f3d-4fae-9dc4-c4c2340316ab",
   "metadata": {},
   "source": [
    "### Run calculating π serially"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1fe8858c-d292-468d-a78a-5aa9e8749721",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Running {NUM_SAMPLING_TASKS} tasks serially....\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fdfd4fb-c4e1-474a-9304-c927eab90471",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "results = run_serial(NUM_SAMPLES_PER_TASK)\n",
    "pi = calculate_pi(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c37f6540-982e-430e-a97d-8c3a67e2074d",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Estimated value of π is: {pi:5f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31c021f9-1f67-483b-91e6-5d0afbf5319a",
   "metadata": {},
   "source": [
    "### Run calculating π with Ray distributed tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c199b3ce-cfa2-4395-807c-7e67726b12ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "results = run_disributed(NUM_SAMPLES_PER_TASK)\n",
    "pi = calculate_pi(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a6714a1a-5794-43bb-916b-7881ac02d27d",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Estimated value of π is: {pi:5f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed990f02-1fa6-4541-a0ed-27e5e4073027",
   "metadata": {},
   "source": [
    "### Recap\n",
    "With Ray, we see an a speed up 🚅 of **~7X**. \n",
    "But what if we decrease the number of samples? Do we get an accurate represenation of π? Try it for yourself. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf8e82bf-dcf0-4c6f-be7e-be3974edbbad",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "1. We started with sample size as `10_000_000`. Can you decrease to `1_000_000`? \n",
    "2. What happens to value of π with fewer sampling data points?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34ffad74-ca79-48be-a006-480aaf9f99ec",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "47743f1f-b65a-46f9-8e27-cc000c29913f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "NUM_SAMPLES_PER_TASK = 1_000_000"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "07bf14f0-f92e-4499-a9c8-d0be99dc7266",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Example 3:  How to use Ray distributed tasks for image transformation and computation\n",
    "For this example, we will simulate a compute-intensive task by transforming and computing some operations on large high-resolution images. These tasks are not uncommon in image classification in a DNN for training and transposing\n",
    "images. \n",
    "\n",
    "PyTorch `torchvision.transforms` API provides many transformation APIs. We will use a couple here, along with some `numpy` and `torch.tensor` operations. Our tasks will perform the following compute-intensive transformations:\n",
    "\n",
    " 1. Use PIL APIs to [blur the image](https://pillow.readthedocs.io/en/stable/reference/ImageFilter.html) with a filter intensity\n",
    " 2. Use Torchvision random [trivial wide augmentation](https://pytorch.org/vision/stable/generated/torchvision.transforms.TrivialAugmentWide.html#torchvision.transforms.TrivialAugmentWide)\n",
    " 3. Convert images into numpy array and tensors and do numpy and torch tensor operations such as [transpose](https://pytorch.org/docs/stable/generated/torch.transpose.html), element-wise [multiplication](https://pytorch.org/docs/stable/generated/torch.mul.html) with a random integers\n",
    " 4. Do more exponential [tensor power](https://pytorch.org/docs/stable/generated/torch.pow.html) and [multiplication with tensors](https://pytorch.org/docs/stable/generated/torch.mul.html)\n",
    "\n",
    "The goal is to compare execution times running these task serially vs. distributed as a Ray Task.\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/images_for_transformation.png\" width=\"80%\" height=\"30%\"> |\n",
    "|:--|\n",
    "|High resolution images for transformation and computation.|"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "66f96e18-923e-4b94-8b43-f48a8d4d4dc4",
   "metadata": {},
   "outputs": [],
   "source": [
    "import tasks_helper_utils as t_utils"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "394af3c2-f68e-4696-84ce-f9d716c63c30",
   "metadata": {},
   "source": [
    "Define some constants that can be tweaked for experimentation with different batch sizes as part of your exercsie."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "07a19d90-9278-4d8d-82b2-a6c3ff900480",
   "metadata": {},
   "outputs": [],
   "source": [
    "DATA_DIR = Path(os.getcwd() + \"/task_images\")\n",
    "BATCHES = [10, 20, 30, 40, 50]\n",
    "SERIAL_BATCH_TIMES = []\n",
    "DISTRIBUTED_BATCH_TIMES = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e2f9aa8-0899-4e3f-ac28-2146f31339c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a Ray task to transform, augment and do some compute intensive tasks on an image\n",
    "@ray.remote\n",
    "def augment_image_distributed(image_ref: object, fetch_image) -> List[object]:\n",
    "    return t_utils.transform_image(image_ref, fetch_image=fetch_image)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "161098df-f1c1-498e-8256-60efce8f11d6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a function to run these transformation tasks serially, on a single node, single core\n",
    "def run_serially(img_list_refs: List) -> List[Tuple[int, float]]:\n",
    "    transform_results = [t_utils.transform_image(image_ref, fetch_image=True) for image_ref in tqdm.tqdm(img_list_refs)]\n",
    "    return transform_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "beb18a7d-a09f-42fe-a960-0655df05dc75",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define function to run these transformation tasks distributed\n",
    "def run_distributed(img_list_refs:List[object]) ->  List[Tuple[int, float]]:\n",
    "    return ray.get([augment_image_distributed.remote(img, False) for img in tqdm.tqdm(img_list_refs)])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0760d0a7-210a-48cc-b4f5-294dc8251267",
   "metadata": {},
   "source": [
    "Let's download 100 large images, each betwen 5-20 MB+ with high-resolution greater (4000, 3500) pixels. It will only download once."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c37d5a7-bf6d-4b0d-b0a4-63f602beb135",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check if dir exists. If so ignore download.\n",
    "# Just assume we have done from a prior run\n",
    "if not os.path.exists(DATA_DIR):\n",
    "    os.mkdir(DATA_DIR)\n",
    "    print(f\"downloading images ...\")\n",
    "    for url in tqdm.tqdm(t_utils.URLS):\n",
    "        t_utils.download_images(url, DATA_DIR)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9828ce0a-37f6-4bda-8043-ce45c758b695",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fetch the the entire image list\n",
    "image_list = list(DATA_DIR.glob(\"*.jpg\"))\n",
    "image_list[:2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90336873-bfdc-4627-8d9e-01e61bdc83be",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Let's look at some of random images, five for now, we are working with. Nice to be one with the data.\n",
    "t_utils.display_random_images(image_list, n=5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "706299ad-36df-4606-9bd4-a0c0bf7363fa",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Populate our Ray object store\n",
    "Since our images are large, let's put them in the [Ray Distributed object store](https://docs.ray.io/en/latest/ray-core/key-concepts.html#objects). (We will cover Ray shared object store in the next tutorial, so bear with me for now).\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Overview_of_Ray/object_store.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Diagram of workers in worker nodes using `ray.put()` to store values and using `ray.get()` to retrieve them from each node's object store.|"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72a0f0c5-a5b7-45ce-8d37-371e05dd48cd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Place all images into the object store. Since Ray tasks may be disributed \n",
    "# across machines, the DATA_DIR may not be present on a worker. However,\n",
    "#placing them into the Ray distributed objector provides access to any \n",
    "# remote task scheduled on Ray worker\n",
    "    \n",
    "images_list_refs = [t_utils.insert_into_object_store(image) for \n",
    "                  image in image_list]\n",
    "images_list_refs[:2]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93659bf2-ab75-4a97-94e2-bf16e0420d87",
   "metadata": {},
   "source": [
    "### Run serially: each image transformation with a Python function\n",
    "\n",
    "We will iterate through the images with batches of 10 (this can be changed 20 or 25, etc) and process them. To simulate a computer-intensive operation on images, we are doing the tensor transformation and computations described above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "37c09e04-fa04-4c45-b9db-302a0dde3922",
   "metadata": {},
   "outputs": [],
   "source": [
    "for idx in BATCHES:\n",
    "    # Use the index to get N number of URLs to images\n",
    "    image_batch_list_refs = images_list_refs[:idx]\n",
    "    print(f\"\\nRunning {len(image_batch_list_refs)} tasks serially....\")\n",
    "    \n",
    "    # Run each one serially\n",
    "    start = time.perf_counter()\n",
    "    serial_results = run_serially(image_batch_list_refs)\n",
    "    end = time.perf_counter()\n",
    "    elapsed = end - start\n",
    "    \n",
    "    # Keep track of batches, execution times as a Tuple\n",
    "    SERIAL_BATCH_TIMES.append((idx, round(elapsed, 2)))\n",
    "    print(f\"Serial transformations/computations of {len(image_batch_list_refs)} images: {elapsed:.2f} sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a0a9e510-0152-41ae-a2b4-b08e00a5bcac",
   "metadata": {},
   "source": [
    "### Run distributed: each image transformation with a Ray task\n",
    "\n",
    "Let's create a Ray task for an image within each batch and process them. Since our images are large and exsist in the [Ray Distributed object store](https://docs.ray.io/en/latest/ray-core/key-concepts.html#objects), our Ray tasks scheduled on any workder node will have access to them.\n",
    "\n",
    "(We will cover Ray shared object store in the next tutorial, so bear with me for now).\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Overview_of_Ray/object_store.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Diagram of workers in worker nodes using `ray.put()` to store values and using `ray.get()` to retrieve them from each node's object store.|"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1cb12234-e7a1-4172-b896-c7268dd2ef73",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Iterate over batches, launching Ray task for each image within the processing\n",
    "# batch\n",
    "for idx in BATCHES:\n",
    "    image_batch_list_refs = images_list_refs[:idx]\n",
    "    print(f\"\\nRunning {len(image_batch_list_refs)} tasks distributed....\")\n",
    "    \n",
    "    # Run each one serially\n",
    "    start = time.perf_counter()\n",
    "    distributed_results = run_distributed(image_batch_list_refs)\n",
    "    end = time.perf_counter()\n",
    "    elapsed = end - start\n",
    "    \n",
    "     # Keep track of batchs, execution times as a Tuple\n",
    "    DISTRIBUTED_BATCH_TIMES.append((idx, round(elapsed, 2)))\n",
    "    print(f\"Distributed transformations/computations of {len(image_batch_list_refs)} images: {elapsed:.2f} sec\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e0b1e0a-64e6-4e82-906e-5a2fe7649cdb",
   "metadata": {},
   "source": [
    "### Compare and plot the serial vs. distributed computational times"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e7e74c33-c85f-4916-99f8-9de21aebf7f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Print times for each and plot them for comparison\n",
    "print(f\"Serial times & batches     : {SERIAL_BATCH_TIMES}\")\n",
    "print(f\"Distributed times & batches: {DISTRIBUTED_BATCH_TIMES}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "17db3ef2-c1d1-4341-8a12-04b229dae6f0",
   "metadata": {},
   "outputs": [],
   "source": [
    "t_utils.plot_times(BATCHES, SERIAL_BATCH_TIMES, DISTRIBUTED_BATCH_TIMES)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a31bb3a5-57c8-4396-a75c-896c274e1127",
   "metadata": {},
   "source": [
    "### Recap\n",
    "\n",
    "We can clearly observe that the overall execution times by Ray tasks is in order of **3-4x** faster 🚅 than serial. Converting an existing serial compute-intensive Python function is as simple as adding the `ray.remote(...)` operator to your Python function. And Ray will handle all the hard bits: scheduling, execution, scaling, memory management, etc.\n",
    "\n",
    "As you can see the benefits are tangible in execution times with Ray tasks."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0658478a",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Exercise\n",
    "\n",
    "1. Write a Python compute intensive function and convert it into a Ray task: pick some function from your repo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bc6626f9-2887-46ca-8c2c-db7a0173fb66",
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0d4fa0e7",
   "metadata": {},
   "source": [
    "### Homework\n",
    "1. For the Example 3, try different batch sizes, and compare the running times. For example, BATCHES = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]\n",
    "2. Read this blog: [Parallelizing Python Code](https://www.anyscale.com/blog/parallelizing-python-code), and try some examples."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4d8d5d0c-89d9-40cc-9fb3-85e3ddc59f3c",
   "metadata": {},
   "source": [
    "### Next Step\n",
    "\n",
    "Let's move on to the distributed [remote objects lesson](Ray_Core_2_Remote_Objects.ipynb)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08f363a1",
   "metadata": {},
   "source": [
    "### References\n",
    "\n",
    "1. [Modern Parallel and Distributed Python: A Quick Tutorial on Ray](https://towardsdatascience.com/modern-parallel-and-distributed-python-a-quick-tutorial-on-ray-99f8d70369b8) by Robert Nishihara, co-creator of Ray and co-founder Anyscale\n",
    "2. [Ray Core Introduction](https://www.anyscale.com/events/2022/02/03/introduction-to-ray-core-and-its-ecosystem) by Jules S. Damji"
   ]
  }
 ],
 "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.8.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}