user_test.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Ray Data User Testing\n",
    "\n",
    "In this notebook, you will learn how to use Ray Data for distributed model training. Ray Data is used for data loading for model training. You are asked to fill in the missing code to finish the following 3 tasks:\n",
    "\n",
    "- Task 1: Read training data from S3\n",
    "- Task 2: Preprocess training data\n",
    "- Task 3: Run distributed training with 2 GPU workers\n",
    "\n",
    "You can refer to the Ray Data Documentation for user guides and APIs: https://docs.ray.io/en/master/data/data.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Task 1: Read training data from S3\n",
    "\n",
    "In this section, you will read training data from AWS S3 (https://aws.amazon.com/s3/). The training data consists of a list of files in Parquet format (https://parquet.apache.org/). The training data contains images and their labels.\n",
    "\n",
    "Success Criteria for this section:\n",
    "- Successfully create a Ray Dataset to read Parquet files from S3.\n",
    "- Successfully inspect metadata of Ray Dataset to get the names of columns, and the number of images."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Instruction: just run this cell.\n",
    "# Import required dependencies.\n",
    "import numpy as np\n",
    "from torchvision.transforms import Compose, Normalize, ToTensor\n",
    "from tqdm import tqdm\n",
    "import ray\n",
    "import ray.train\n",
    "from ray.train import ScalingConfig\n",
    "from ray.train.torch import TorchTrainer\n",
    "from util import prepare_model\n",
    "\n",
    "path = \"s3://air-example-data/data-cuj/train\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instruction: add your code here.\n",
    "# Read from S3 bucket above: s3://air-example-data/data-cuj/train\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please answer the questions below:\n",
    "\n",
    "- What are the names of columns?\n",
    "\n",
    "\n",
    "- How many images in total?\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Task 2: Preprocess images\n",
    "\n",
    "In this section, we'll preprocess images to normalize them for training. You are given the preprocess code below to normalize a single image with TorchVision function (`transform()`). You are expected to use the Ray Data API to parallelize the preprocessing logic among all images. \n",
    "\n",
    "Success Criteria for this section:\n",
    "- Successfully use correct Ray Data API to run preprocessing for all images.\n",
    "- Successfully inspect the first row of preprocessed Dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Instruction: just run this cell.\n",
    "# The preprocessing function to be applied to every image.\n",
    "transform = Compose([ToTensor(), Normalize((0.5,), (0.5,))])\n",
    "\n",
    "# Example of running the preprocessing on a toy image.\n",
    "example_image = np.array([[1, 0, 1], [1, 1, 0.5]], np.double)\n",
    "transform(example_image)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "NOTE:\n",
    "- The above preprocessing function (`transform()`) only takes `image` as input.\n",
    "- The input Dataset has two columns - `image` and `label`. Do not drop `label` after applying preprocessing!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Instruction: add your code here.\n",
    "# Use Ray Data to parallelize the preprocessing logic above.\n",
    "\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please answer the question below:\n",
    "\n",
    "- What is the first row of Dataset?\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Task 3: Run PyTorch distributed training with 2 GPU workers\n",
    "\n",
    "In this section, let's do distributed PyTorch model training for preprocessed images. To do distributed model training, you are given a script with everything working, except the part of data loading is missing. You are expected to fill in the data loading code and start training!\n",
    "\n",
    "Success Criteria for this section:\n",
    "- Successfully use correct Ray Data API to get shard of Dataset.\n",
    "- Successfully use correct Ray Data API to iterate the Dataset and pass batch of (images, labels) to model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# The function executed on each training worker.\n",
    "def train_func_per_worker():\n",
    "    # Prepare model.\n",
    "    model, loss_fn, optimizer = prepare_model()\n",
    "\n",
    "    # Model training for 2 epochs.\n",
    "    for epoch in range(2):\n",
    "        # Instruction: add your code here to get Dataset.\n",
    "        # NOTE: each training data batch should have 32 images and labels.\n",
    "        # ...\n",
    "        # train_dataloader = ...\n",
    "\n",
    "        model.train()\n",
    "        # Model training for each batch.\n",
    "        for batch in tqdm(train_dataloader, desc=f\"Train Epoch {epoch}\"):\n",
    "            images = batch[\"image\"]\n",
    "            labels = batch[\"label\"]\n",
    "            pred = model(images)\n",
    "            loss = loss_fn(pred, labels)\n",
    "\n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "\n",
    "        model.eval()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Initialize a Ray TorchTrainer\n",
    "trainer = TorchTrainer(\n",
    "    train_loop_per_worker=train_func_per_worker,\n",
    "    scaling_config=ScalingConfig(num_workers=2, use_gpu=True),\n",
    "    # Instruction: add your code here to pass the Dataset variable.\n",
    "    datasets={\"train\": ...}\n",
    ")\n",
    "\n",
    "# Start model training!\n",
    "result = trainer.fit()\n",
    "print(f\"Training result: {result}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please answer the questions below:\n",
    "\n",
    "- How many epochs have been trained?\n",
    "- What is end-to-end training time (in seconds)?"
   ]
  }
 ],
 "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.11.0 (main, Oct 25 2022, 14:13:24) [Clang 14.0.0 (clang-1400.0.29.202)]"
  },
  "vscode": {
   "interpreter": {
    "hash": "1a1af0ee75eeea9e2e1ee996c87e7a2b11a0bebd85af04bb136d915cefc0abce"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}