ex_02_remo_func.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a4ad1455-57a4-439f-aeb8-d58d8b48fb89",
   "metadata": {},
   "source": [
    "# A Guided Tour of Ray Core: Remote Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c50a00c6-a6d4-48fc-a823-4d3d10417912",
   "metadata": {},
   "source": [
    "[*Remote Functions*](https://docs.ray.io/en/latest/walkthrough.html#remote-functions-tasks)\n",
    "involve using a `@ray.remote` decorator on a function. \n",
    "\n",
    "This implements a [*task parallelism*](https://patterns.eecs.berkeley.edu/?page_id=208) pattern, with properties: *data independence*, *stateless*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "902059f3-959a-418e-a8d8-670fb11fff5a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from numpy import loadtxt\n",
    "import ray"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d968d45e-7162-4116-834d-a4f9a29598c6",
   "metadata": {},
   "source": [
    "Ray converts decorated functions into stateless tasks, scheduled anywhere onto a ray worker in the cluster.\n",
    "\n",
    "This remote task reads a file and returns its contents as a numpy array."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ddb7b82-1b20-4bbd-96e0-6b9e98afd1e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def read_array(fn: str) -> np.array:\n",
    "    arr = loadtxt(fn, comments=\"#\", delimiter=\",\", unpack=False)\n",
    "    return arr.astype('int')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fe1111fe-5b85-4a84-969c-ea5402c7f1a4",
   "metadata": {},
   "source": [
    "Given two numpy arrays, this remote task returns add the two numpy arrays"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c509fad-96dd-473c-bec3-c5a6bbce64c8",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def add_array(a1: np.array, a2: np.array) -> np.array:\n",
    "    return np.add(a1, a2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b6283055-3ef4-4d6e-9959-4ffc050e9a4d",
   "metadata": {},
   "source": [
    "Given a numpy array, add its contents"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9de6e60c-867d-43d0-b913-f4044c4e5d11",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def sum_arr(a1: np.array) -> int:\n",
    "    return a1.sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57f40e79-53f0-4dac-a703-cb9e303c9a76",
   "metadata": {},
   "source": [
    "Ray executes all remote tasks on an assigned node and returns future immediately. Futures enable asynchonrous calls, which enables concurrency and parallelism.\n",
    "\n",
    "**Note**: I did not start the Ray iwth `ray.init(..)` as in previous excercise; the first call to Ray, automatically launched \n",
    "a Ray on my local host. Since this is a local mode, I'll only have a headnode running."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ac942a0-2d9b-4fd1-a8ec-8f1cedb23b47",
   "metadata": {},
   "outputs": [],
   "source": [
    "obj_ref_arr1 = read_array.remote(\"dat/file_1.txt\")\n",
    "print(f\"array 1: {obj_ref_arr1}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a69772fc-5808-4d20-9ce0-b4006e587d63",
   "metadata": {},
   "outputs": [],
   "source": [
    "obj_ref_arr2 = read_array.remote(\"dat/file_2.txt\")\n",
    "print(f\"array 2: {obj_ref_arr2}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e4204e60-16ca-4b49-bb30-4e1561253e03",
   "metadata": {},
   "source": [
    "Ray executes the remote task to add two arrays stored in the object references and immediately \n",
    "returns an object reference. \n",
    "\n",
    "**Note**: That you can send object references or futures to the function. Ray will resolve it at the time\n",
    "of executing the task, to resolve them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ae81b41-7a5a-487b-87ba-dcf688d6ff06",
   "metadata": {},
   "outputs": [],
   "source": [
    "result_obj_ref = add_array.remote(obj_ref_arr1, obj_ref_arr2)\n",
    "result_obj_ref"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b74801f-fdf3-47f5-abb8-35e2fd5e6e7b",
   "metadata": {},
   "source": [
    "Fetch the result: this will block if not finished"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c611019-dcfd-465e-9d60-97c48cd3c92b",
   "metadata": {},
   "outputs": [],
   "source": [
    "result = ray.get(result_obj_ref)\n",
    "print(f\"Result: add arr1 + arr2: {result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "209c4756-ea6a-47c9-bcf9-8e28493059de",
   "metadata": {},
   "source": [
    "Add the array elements in each numpy array and get the sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "de2b1350-2d29-4a31-a0d3-b307e5c0f646",
   "metadata": {},
   "outputs": [],
   "source": [
    "sum_1 = ray.get(sum_arr.remote(obj_ref_arr1))\n",
    "sum_2 = ray.get(sum_arr.remote(obj_ref_arr2))\n",
    "\n",
    "print(f'Sum of arr1: {sum_1}')\n",
    "print(f'Sum of arr2: {sum_2}')"
   ]
  }
 ],
 "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.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}