ex_04_remote_classes_revisited.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Guided Tour of Ray Core: Ray Actor Tree Pattern\n",
    "\n",
    "© 2019-2022, Anyscale. All Rights Reserved\n",
    "\n",
    "📖 [Back to Table of Contents](./ex_00_tutorial_overview.ipynb)<br>\n",
    "➡ [Next notebook](./ex_05_multiprocess_pool.ipynb) <br>\n",
    "⬅️ [Previous notebook](./ex_03_remote_classes.ipynb) <br>\n",
    "\n",
    "### Overview\n",
    "\n",
    "Ray has a myriad of design patterns for [tasks](https://docs.ray.io/en/latest/ray-core/tasks/patterns/index.html#task-patterns) and [actors](https://docs.ray.io/en/latest/ray-core/actors/patterns/index.html). These patterns allows you to write distributed applications. In this tutorial we'll explore one of the design pattern, commonly used in Ray libraries to scale workloads.\n",
    "\n",
    "#### Tree of Actors Pattern\n",
    "\n",
    "This pattern is primarily used in Ray libraries [Ray Tune](https://docs.ray.io/en/latest/tune/index.html), [Ray Train](https://docs.ray.io/en/latest/train/train.html), and [RLlib](https://docs.ray.io/en/latest/rllib/index.html) to train models in a parallel or conduct distributed HPO.\n",
    "\n",
    "In a tree of actors pattern, a collection of workers as Ray actors (or in some cases Ray tasks too), are managed by a supervisor actor. For example, you want to train multiple models, each of a different ML algorithm, at the same time, while being able to inspect its state during its training. As methology to examine simple baseline models rapidly, this pattern helps machine learning engineers to quickly build a set of baseline models for comparison.\n",
    "\n",
    "|<img src=\"https://docs.ray.io/en/latest/_images/tree-of-actors.svg\" width=\"85%\" height=\"45%\">|\n",
    "|:--|\n",
    "| Common tree actor pattern with a supervisor launching a supervisor actor|\n",
    "\n",
    "This pattern facilitates **Same Data Different Function/Model (SDDF)** scaling pattern. Popular in [AutoML](https://en.wikipedia.org/wiki/Automated_machine_learning) scenarios, where you may want to train different models, with their respective algorithms, at the same time using the same dataset.\n",
    "\n",
    "|<img src=\"images/same_data_different_model.png\" width=\"85%\" height=\"45%\">|\n",
    "|:--|\n",
    "| Same data different function/model is a common scaling pattern|\n",
    "\n",
    "### Learning objectives\n",
    "In this this tutorial, we revisit Ray Actors, diving deeper its use for scaling patterns, and learn more about:\n",
    " * Common Ray Actors patterns used in Ray native libraries for writing distributed Actors\n",
    "   * Tree of Actors \n",
    "   * Same Data Different Functions (SDDF)\n",
    " * How to use Actors and ActorPool for Batch Inference\n",
    "\n",
    "Let's implement a simple example to illustrate this pattern."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some preliminary imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "import time\n",
    "import ray\n",
    "import random\n",
    "from random import randint\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import pyarrow.parquet as pq\n",
    "from pprint import pprint\n",
    "\n",
    "import ray\n",
    "from ray.util.actor_pool import ActorPool"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, let's start Ray…"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "    <div style=\"margin-left: 50px;display: flex;flex-direction: row;align-items: center\">\n",
       "        <h3 style=\"color: var(--jp-ui-font-color0)\">Ray</h3>\n",
       "        <svg version=\"1.1\" id=\"ray\" width=\"3em\" viewBox=\"0 0 144.5 144.6\" style=\"margin-left: 3em;margin-right: 3em\">\n",
       "            <g id=\"layer-1\">\n",
       "                <path fill=\"#00a2e9\" class=\"st0\" d=\"M97.3,77.2c-3.8-1.1-6.2,0.9-8.3,5.1c-3.5,6.8-9.9,9.9-17.4,9.6S58,88.1,54.8,81.2c-1.4-3-3-4-6.3-4.1\n",
       "                    c-5.6-0.1-9.9,0.1-13.1,6.4c-3.8,7.6-13.6,10.2-21.8,7.6C5.2,88.4-0.4,80.5,0,71.7c0.1-8.4,5.7-15.8,13.8-18.2\n",
       "                    c8.4-2.6,17.5,0.7,22.3,8c1.3,1.9,1.3,5.2,3.6,5.6c3.9,0.6,8,0.2,12,0.2c1.8,0,1.9-1.6,2.4-2.8c3.5-7.8,9.7-11.8,18-11.9\n",
       "                    c8.2-0.1,14.4,3.9,17.8,11.4c1.3,2.8,2.9,3.6,5.7,3.3c1-0.1,2,0.1,3,0c2.8-0.5,6.4,1.7,8.1-2.7s-2.3-5.5-4.1-7.5\n",
       "                    c-5.1-5.7-10.9-10.8-16.1-16.3C84,38,81.9,37.1,78,38.3C66.7,42,56.2,35.7,53,24.1C50.3,14,57.3,2.8,67.7,0.5\n",
       "                    C78.4-2,89,4.7,91.5,15.3c0.1,0.3,0.1,0.5,0.2,0.8c0.7,3.4,0.7,6.9-0.8,9.8c-1.7,3.2-0.8,5,1.5,7.2c6.7,6.5,13.3,13,19.8,19.7\n",
       "                    c1.8,1.8,3,2.1,5.5,1.2c9.1-3.4,17.9-0.6,23.4,7c4.8,6.9,4.6,16.1-0.4,22.9c-5.4,7.2-14.2,9.9-23.1,6.5c-2.3-0.9-3.5-0.6-5.1,1.1\n",
       "                    c-6.7,6.9-13.6,13.7-20.5,20.4c-1.8,1.8-2.5,3.2-1.4,5.9c3.5,8.7,0.3,18.6-7.7,23.6c-7.9,5-18.2,3.8-24.8-2.9\n",
       "                    c-6.4-6.4-7.4-16.2-2.5-24.3c4.9-7.8,14.5-11,23.1-7.8c3,1.1,4.7,0.5,6.9-1.7C91.7,98.4,98,92.3,104.2,86c1.6-1.6,4.1-2.7,2.6-6.2\n",
       "                    c-1.4-3.3-3.8-2.5-6.2-2.6C99.8,77.2,98.9,77.2,97.3,77.2z M72.1,29.7c5.5,0.1,9.9-4.3,10-9.8c0-0.1,0-0.2,0-0.3\n",
       "                    C81.8,14,77,9.8,71.5,10.2c-5,0.3-9,4.2-9.3,9.2c-0.2,5.5,4,10.1,9.5,10.3C71.8,29.7,72,29.7,72.1,29.7z M72.3,62.3\n",
       "                    c-5.4-0.1-9.9,4.2-10.1,9.7c0,0.2,0,0.3,0,0.5c0.2,5.4,4.5,9.7,9.9,10c5.1,0.1,9.9-4.7,10.1-9.8c0.2-5.5-4-10-9.5-10.3\n",
       "                    C72.6,62.3,72.4,62.3,72.3,62.3z M115,72.5c0.1,5.4,4.5,9.7,9.8,9.9c5.6-0.2,10-4.8,10-10.4c-0.2-5.4-4.6-9.7-10-9.7\n",
       "                    c-5.3-0.1-9.8,4.2-9.9,9.5C115,72.1,115,72.3,115,72.5z M19.5,62.3c-5.4,0.1-9.8,4.4-10,9.8c-0.1,5.1,5.2,10.4,10.2,10.3\n",
       "                    c5.6-0.2,10-4.9,9.8-10.5c-0.1-5.4-4.5-9.7-9.9-9.6C19.6,62.3,19.5,62.3,19.5,62.3z M71.8,134.6c5.9,0.2,10.3-3.9,10.4-9.6\n",
       "                    c0.5-5.5-3.6-10.4-9.1-10.8c-5.5-0.5-10.4,3.6-10.8,9.1c0,0.5,0,0.9,0,1.4c-0.2,5.3,4,9.8,9.3,10\n",
       "                    C71.6,134.6,71.7,134.6,71.8,134.6z\"/>\n",
       "            </g>\n",
       "        </svg>\n",
       "        <table>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left\"><b>Python version:</b></td>\n",
       "                <td style=\"text-align: left\"><b>3.8.13</b></td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "                <td style=\"text-align: left\"><b>Ray version:</b></td>\n",
       "                <td style=\"text-align: left\"><b> 2.2.0</b></td>\n",
       "            </tr>\n",
       "            <tr>\n",
       "    <td style=\"text-align: left\"><b>Dashboard:</b></td>\n",
       "    <td style=\"text-align: left\"><b><a href=\"http://127.0.0.1:8265\" target=\"_blank\">http://127.0.0.1:8265</a></b></td>\n",
       "</tr>\n",
       "\n",
       "        </table>\n",
       "    </div>\n",
       "</div>\n"
      ],
      "text/plain": [
       "RayContext(dashboard_url='127.0.0.1:8265', python_version='3.8.13', ray_version='2.2.0', ray_commit='b6af0887ee5f2e460202133791ad941a41f15beb', address_info={'node_ip_address': '127.0.0.1', 'raylet_ip_address': '127.0.0.1', 'redis_address': None, 'object_store_address': '/tmp/ray/session_2022-12-21_18-50-48_449739_8819/sockets/plasma_store', 'raylet_socket_name': '/tmp/ray/session_2022-12-21_18-50-48_449739_8819/sockets/raylet', 'webui_url': '127.0.0.1:8265', 'session_dir': '/tmp/ray/session_2022-12-21_18-50-48_449739_8819', 'metrics_export_port': 52203, 'gcs_address': '127.0.0.1:64010', 'address': '127.0.0.1:64010', 'dashboard_agent_listen_port': 52365, 'node_id': '1d548d14c7671b76118a397077bfd53ca11f2bc7343fe562866a00b7'})"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "if ray.is_initialized:\n",
    "    ray.shutdown()\n",
    "ray.init(logging_level=logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example 1: Supervisor and worker actor pattern\n",
    "\n",
    "**Problem**: We want to train multiple small models, say of type linear regression, each with their respective machine learning algorithm, on the same dataset, to create a set of baseline models for comparison. In our case, we use the scikit-learn [California house price](https://scikit-learn.org/stable/modules/generated/sklearn.datasets.fetch_california_housing.html) dataset and use models of type linear regression. We'll train different linear regression models: Decision tree, random forest, and xgboost regressor.  Each model is trained and evaluated based on [mean square error](https://en.wikipedia.org/wiki/Mean_squared_error)(MSE).\n",
    "\n",
    "To see this pattern used in production at scale, read [Training One Million Machine Learning Models in Record Time with Ray](https://www.anyscale.com/blog/training-one-million-machine-learning-models-in-record-time-with-ray)\n",
    "\n",
    "**NOTE**: In our second [Ray core tutorial](ex_02_remote_objs.ipynb), we used one ML algorithm, an ensemble [Random Forest Regressor](https://scikit-learn.org/stable/modules/generated/sklearn.ensemble.RandomForestRegressor.html) using the same dataset but with many  tasks, each with a with different hyperparameter, namely estimators, to gauge training time and MSE accuracy. \n",
    "\n",
    "|<img src=\"images/California_dataset.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|`n_samples = 20640`, target is numeric and corresponds to the average house value in units of 100k.|\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generic model factory utility  \n",
    "\n",
    "This factory generates a three different algorithms for linear regression–random forest, decsion tree, and xgboost–and trains on the same Scikit learn dataset [California housing price](https://scikit-learn.org/stable/modules/generated/sklearn.datasets.fetch_california_housing.html)\n",
    "Each training model returns its MSE score, along with time to train and relevant parameters. \n",
    "\n",
    "Each model can be in a particular state during training. The final state is `DONE`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from model_helper_utils import RFRActor\n",
    "from model_helper_utils import DTActor\n",
    "from model_helper_utils import XGBoostActor\n",
    "from model_helper_utils import RANDOM_FOREST_CONFIGS, DECISION_TREE_CONFIGS, XGBOOST_CONFIGS\n",
    "\n",
    "class ModelFactory:\n",
    "    \"\"\"\n",
    "    Model factory to create different ML models\n",
    "    \"\"\"\n",
    "    MODEL_TYPES = [\"random_forest\", \"decision_tree\", \"xgboost\"]\n",
    "    \n",
    "    @staticmethod\n",
    "    def create_model(model_name: str) -> ray.actor.ActorHandle:\n",
    "        if model_name not in ModelFactory.MODEL_TYPES:\n",
    "            raise Exception(f\"{model_name} not supported\")\n",
    "        if model_name == \"random_forest\":\n",
    "            configs = RANDOM_FOREST_CONFIGS\n",
    "            return RFRActor.remote(configs)\n",
    "        elif model_name == \"decision_tree\":\n",
    "            configs = DECISION_TREE_CONFIGS\n",
    "            return DTActor.remote(configs)\n",
    "        else: \n",
    "            configs = XGBOOST_CONFIGS\n",
    "            return XGBoostActor.remote(configs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Create an Actor instance as a supervisor\n",
    "The supervisor creates three worker actors, each with its own respective training model ML algorithm and its training function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class Supervisor:\n",
    "    def __init__(self):\n",
    "        # Create three Actor Workers, each by its unique model type and \n",
    "        # their respective training function\n",
    "        self.worker_models =  [ModelFactory.create_model(name) for name in ModelFactory.MODEL_TYPES]\n",
    "\n",
    "    def work(self):\n",
    "        # do the train work for each Actor model\n",
    "        results = [worker_model.train_and_evaluate_model.remote() for worker_model in self.worker_models]\n",
    "        \n",
    "        # Return the final results\n",
    "        return ray.get(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Launch Supervisor's workers\n",
    " * create the Supervisor actor\n",
    " * launch its workers\n",
    " * fetch the final results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[2m\u001b[36m(XGBoostActor pid=8874)\u001b[0m Start training model xgboost with estimators: 150 and max depth: 10 ...\n",
      "\u001b[2m\u001b[36m(DTActor pid=8873)\u001b[0m Start training model decision_tree with max depth: 15 ...\n",
      "\u001b[2m\u001b[36m(RFRActor pid=8872)\u001b[0m Start training model random_forest with estimators: 150 ...\n",
      "\u001b[2m\u001b[36m(DTActor pid=8873)\u001b[0m End training model decision_tree with max_depth tree: 15 took: 0.09 seconds\n",
      "\u001b[2m\u001b[36m(XGBoostActor pid=8874)\u001b[0m End training model xgboost with estimators: 150 and max depth: 10 and took: 3.48\n"
     ]
    }
   ],
   "source": [
    "supervisor = Supervisor.remote()\n",
    "results = supervisor.work.remote()\n",
    "values = ray.get(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let for the workers to finish by returning their `DONE` state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "states = []\n",
    "# Wait for all models to finish\n",
    "while True:\n",
    "    for value in values:\n",
    "        states.append(value[\"state\"])\n",
    "    result = all('DONE' == e for e in states)\n",
    "    if result:\n",
    "        break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Results from three training models sorted by MSE ascending order:\n",
      "[{'estimators': 150,\n",
      "  'max_depth': 10,\n",
      "  'mse': 0.2116,\n",
      "  'name': 'xgboost',\n",
      "  'state': 'DONE',\n",
      "  'time': 3.48},\n",
      " {'estimators': 150,\n",
      "  'mse': 0.2529,\n",
      "  'name': 'random_forest',\n",
      "  'state': 'DONE',\n",
      "  'time': 9.48},\n",
      " {'max_depth': 15,\n",
      "  'mse': 0.4859,\n",
      "  'name': 'decision_tree',\n",
      "  'state': 'DONE',\n",
      "  'time': 0.09}]\n"
     ]
    }
   ],
   "source": [
    "from operator import itemgetter\n",
    "sorted_by_mse = sorted(values, key=itemgetter('mse'))\n",
    "print(f\"\\nResults from three training models sorted by MSE ascending order:\")\n",
    "pprint(sorted_by_mse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Look at the Ray Dashboard\n",
    "\n",
    "You should see Actors running as process on the workers nodes\n",
    " * Supervisor\n",
    " * Eeach model actor: RFRActor, DTActor, XGBoostActor\n",
    " \n",
    "Also, click on the `Actors` to view more metrics and data on individual Ray Actors"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap\n",
    "To sum up, we went through implementing a Tree actor design pattern, in which a Supervisor\n",
    "launched worker models. Each model, in our case, is a separate linear regression model training on the same data. This tree actor pattern facilitates the **Same Data Different Functions (SDDF)** scaling pattern, a common machine learning workload.\n",
    "\n",
    "For modularity, we used `model_helper_utils`, where all the model-related utility code is implemented. From all three linear regression models trained in parallel, with the same dataset, XGBoost seems to have faired well with the best MSE result. This gaves us an initial baseline of set of models to further experiment with different [hyperparameter optimization](https://en.wikipedia.org/wiki/Hyperparameter_optimization). For that I would turn to [Ray Train](https://docs.ray.io/en/latest/train/train.html) and [Ray Tune](https://docs.ray.io/en/latest/tune/index.html), whhich are part of [Ray AIR](https://docs.ray.io/en/latest/ray-air/getting-started.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge Exercise (Optional):\n",
    "\n",
    "Add another linear regression model and train four models. \n",
    "\n",
    "**HINTS**:\n",
    " * modify `model_helper_utils` to create a new super class of `ActorCls`\n",
    " * add new model configs for the new model\n",
    " * modify the `ModelFactory` class instantiate this new model\n",
    " * update the `Supervisor` to include the new model in its training\n",
    " \n",
    " You can use one of the linear models from this [array of linear models](https://www.kaggle.com/code/faressayah/practical-introduction-to-10-regression-algorithm).\n",
    " \n",
    "Have fun 😜!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example 2: Actor-based batch inference\n",
    "\n",
    "In our first tutorial on [Ray remote objects](./ex_02_remote_objs.ipynb) we covered a use case to parallelize batch inference. In short, we\n",
    "used the **Different Data Same Function(DDSF)** pattern. Here we will use the same pattern but with Ray Actors and [ActorPool](https://docs.ray.io/en/latest/ray-core/actors/actor-utils.html?highlight=ActorPool), which are state-based and preferred method to do batch inference. Primarily, the elements remain the same except for few modifications.\n",
    "\n",
    "**Input dataset**: This is a large collection of observations to generate predictions for. The data is usually stored in an external storage system like S3, HDFS or database, across\n",
    "many, files.\n",
    "\n",
    "**ML model**: This is a trained ML model that is usually also stored in an external storage system or in a model store.\n",
    "\n",
    "**Predictions**: These are the outputs when applying the ML model on observations. Normally, predictions are usually written back to the storage system. Unlike tasks\n",
    "doing the predictions, we employ a pool of Actors.\n",
    "\n",
    "For purpose of this tutorial, we make the following provisions:\n",
    " * create a dummy model that returns some fake prediction\n",
    " * use real-world NYC taxi data to provide large data set for batch inference\n",
    " * create a pool of actors and submit each shard to the pool.\n",
    " * return the predictions instead of writing it back to the disk\n",
    "\n",
    "As an example of scaling pattern called **Different Data Same Function (DDSF)**, also known as **Distributed Data Parallel** (DDP) paradigm, our function in this digaram is the \n",
    "pretrained **model** and the data is split and disributed as **shards**.\n",
    "\n",
    "|<img src=\"images/batch-inference.png\" width=\"75%\" height=\"35%\">|\n",
    "|:--|\n",
    "|Different data same function is another scaling pattern|\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "NUM_ACTORS = 6             # You can always increase the number of actors to scale\n",
    "NUM_SHARD_FILES = 12       # number of shard files you want each each actor to work on"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Our load model closure remains the same\n",
    "def load_trained_model():\n",
    "    # A fake model that predicts whether tips were given based on number of passengers in the taxi cab.\n",
    "    def model(batch: pd.DataFrame) -> pd.DataFrame:\n",
    "        # Some model payload so Ray copies the model in the shared plasma store to tasks scheduled across nodes.\n",
    "        model.payload = np.arange(100, 100_000_000, dtype=float)\n",
    "        model.cls = \"regression\"\n",
    "        \n",
    "        # give a tip if 2 or more passengers\n",
    "        predict = batch[\"passenger_count\"] >= 2 \n",
    "        return pd.DataFrame({\"score\": predict})\n",
    "    \n",
    "    return model    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a Ray Actor that stores a model reference and does the prediction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class NYCBatchPredictor:\n",
    "    def __init__(self, model):\n",
    "        self.model = model\n",
    "\n",
    "    def predict(self, shard_path):\n",
    "        # read each shard and convert to pandas\n",
    "        df = pq.read_table(shard_path).to_pandas()\n",
    "        \n",
    "        # do the inference with our model and return the result\n",
    "        result = self.model(df)\n",
    "        return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. Get our trained model instance\n",
    "2. Store it into the plasma object store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = load_trained_model()\n",
    "model_ref = ray.put(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Fetch our NYC taxi shard files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Iterate thorough our NYC files ~ 2GB\n",
    "input_shard_files = [\n",
    "    f\"s3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet\"\n",
    "    f\"/fe41422b01c04169af2a65a83b753e0f_{i:06d}.parquet\"\n",
    "    for i in range(NUM_SHARD_FILES) ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['s3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000000.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000001.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000002.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000003.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000004.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000005.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000006.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000007.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000008.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000009.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000010.parquet',\n",
       " 's3://anonymous@air-example-data/ursa-labs-taxi-data/downsampled_2009_full_year_data.parquet/fe41422b01c04169af2a65a83b753e0f_000011.parquet']"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "input_shard_files"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. Create five Actor instances, each initialized with the same model reference\n",
    "2. Create a pool of five actors\n",
    "\n",
    "We use the Ray actor pool utility [ActorPool](https://docs.ray.io/en/latest/ray-core/actors/actor-utils.html?highlight=ActorPool#actor-pool).\n",
    "\n",
    "[Actool Pool API](https://docs.ray.io/en/latest/ray-core/package-ref.html?highlight=ActorPool#ray-util-actorpool) reference package."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "actors = [NYCBatchPredictor.remote(model_ref) for _ in range(NUM_ACTORS)]\n",
    "actors_pool = ActorPool(actors)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Submit each shard to the pool of actors for batch reference\n",
    "# The API syntax is not dissimilar to Python or Ray Multiprocessor pool APIs\n",
    "\n",
    "for shard_path in input_shard_files:\n",
    "    # Submit file shard for prediction to the pool\n",
    "    actors_pool.submit(lambda actor, shard: actor.predict.remote(shard), shard_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[2m\u001b[36m(RFRActor pid=8872)\u001b[0m End training model random_forest with estimators: 150 took: 9.48 seconds\n",
      "Predictions dataframe size: 141062 | Total score for tips: 46360\n",
      "Predictions dataframe size: 133932 | Total score for tips: 42175\n",
      "Predictions dataframe size: 144014 | Total score for tips: 45175\n",
      "Predictions dataframe size: 143087 | Total score for tips: 45510\n",
      "Predictions dataframe size: 148108 | Total score for tips: 47713\n",
      "Predictions dataframe size: 141981 | Total score for tips: 45188\n",
      "Predictions dataframe size: 136394 | Total score for tips: 43234\n",
      "Predictions dataframe size: 136999 | Total score for tips: 45142\n",
      "Predictions dataframe size: 139985 | Total score for tips: 44138\n",
      "Predictions dataframe size: 156198 | Total score for tips: 49909\n",
      "Predictions dataframe size: 142893 | Total score for tips: 46112\n",
      "Predictions dataframe size: 145976 | Total score for tips: 48036\n"
     ]
    }
   ],
   "source": [
    "# Iterate over finised actor's predictions\n",
    "while actors_pool.has_next():\n",
    "    r =  actors_pool.get_next()\n",
    "    print(f\"Predictions dataframe size: {len(r)} | Total score for tips: {r['score'].sum()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap\n",
    "\n",
    "What we have demonstrated above is an Actor tree design pattern, commonly used in Ray for writing distributed applications. In particular, Ray's native libraries such as Train, Tune, Serve, and RLib and [Ray AIR's](https://docs.ray.io/en/latest/ray-air/getting-started.html) components use it for distributed training and tuning trials. \n",
    "\n",
    "Additionally, we implemented a DDSF scaling design pattern using an Actor-based predictor model function, using an `ActorPool` utility class instead of task.  \n",
    "\n",
    "Task-based batch inferene has an overhead cost that can be significant if the model size is large, since it has to fetch the model from the driver's plasma store. We can optimize it by using Ray actors, \n",
    "which will fetch the model just once and reuse it for all predictions assigned to the same actor in the pool."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Next Step\n",
    "We going to switch a focus little and learn how you can use Ray's replacement\n",
    "for Python's Multiprocessing pool. \n",
    "\n",
    "Let's move on to the [Multiprocessing pool with Ray](ex_05_multiprocess_pool.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Homework\n",
    "\n",
    "1. Read references below."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## References"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " * [Writing your First Distributed Python Application with Ray](https://www.anyscale.com/blog/writing-your-first-distributed-python-application-with-ray)\n",
    " * [Using and Programming with Actors](https://docs.ray.io/en/latest/actors.html)\n",
    " * [Ray Asynchronous and Threaded Actors: A way to achieve concurrency](https://medium.com/@2twitme/ray-asynchronous-and-threaded-actors-a-way-to-achieve-concurrency-ad9f86145f72)\n",
    " * [Training One Million Machine Learning Models in Record Time with Ray](https://www.anyscale.com/blog/training-one-million-machine-learning-models-in-record-time-with-ray)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "📖 [Back to Table of Contents](./ex_00_tutorial_overview.ipynb)<br>\n",
    "➡ [Next notebook](./ex_05_multiprocess_pool.ipynb) <br>\n",
    "⬅️ [Previous notebook](./ex_03_remote_classes.ipynb) <br>"
   ]
  }
 ],
 "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": 4
}