model_training_full_sequence.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MNIST Handwritten Digit Recognition using Astra and Pytorch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Import the packages requires for the first step of Data Loader setup.\n",
    "\n",
    "# From pip install cassandra-driver, necessary to connect to AstraDB, general rather than specific implementations used here\n",
    "from cassandra.cluster import Cluster\n",
    "from cassandra.auth import PlainTextAuthProvider\n",
    "\n",
    "# From pip install torch we need the Dataset and DataLoader types to create a loader that fetchees data from Astra in a form compatible with Pytorch\n",
    "import torch\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "\n",
    "# From pip install torchvision, provides the transforms necesarry to make raw pixel values from Astra into proper image format for machine learning\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# Already installed in Python environment, essentially matrix processing faster than python Arrays\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Astra Pytorch Dataset Definition\n",
    "\n",
    "# First we create a class AstraDataset Based on the existing Pytorch Dataset class. Datasets define a method for pulling particular subsets of larger data sources, \n",
    "# whether those are internal python objects, files, or external data sources like AstraDB.\n",
    "class AstraDataset(Dataset):\n",
    "    # In the init function we describe what arguments are necesarry to create an AstraDataset and how those become it's  internal variables.\n",
    "    # * cloud_conifg: Specifies the location of the Cassandra database to connect to. Here we essentially pass the secure connect bundle, which contains the necesarry data.\n",
    "    # * auth_provider: Provides the authentication nmecesarry to connect to the specified database. We create a PlainTextAuthProvider object containing the id and secret from Astra that\n",
    "    # ** are associated with the Database Adminitrator token we generate.\n",
    "    # * keyspace: The keyspace of the table that we want to pull data from\n",
    "    # * table: The name of the table that we want to pull data from.\n",
    "    # * length: How much data we want to be in this particular AstraDataset. This pulls objects based on an id number in order from 0.\n",
    "    # * transform: The transform function to be applied to data after it is pulled in from Astra\n",
    "    def __init__(self,\n",
    "                cloud_config={},\n",
    "                auth_provider=None,\n",
    "                keyspace=\"\",\n",
    "                table=\"raw_train\",\n",
    "                length=0,\n",
    "                transform=None):\n",
    "        # Here we create the session object corresponding to the Astra database from the components provided above\n",
    "        self.db = Cluster(cloud=cloud_config, auth_provider=auth_provider).connect()\n",
    "        self.keyspace = keyspace\n",
    "        self.table = table\n",
    "        self.length = length\n",
    "        self.transform = transform\n",
    "\n",
    "    # The get item function pulls data from the partiuclar row, separates it into data and label, and applies the transform to the data section.\n",
    "    def __getitem__(self, index):\n",
    "        # Effective query is SELECT pixels from keyspace.table WHERE id = index\n",
    "        # Then .one pulls the single row from the results (since index is the full primary key, no chance of getting more than one row per index)\n",
    "        # Then [0] pulls out the only column pixels, which is an array of 784 grayscale pixel darkness values from 0 at pure black to 255 at pure white\n",
    "        # Then we reshape the 1D array into a 2D 28 by 28 image and divide by 255 so the pixel values then range from 0 being pure black and 1 being pure white\n",
    "        # Then we apply the specifieds transforms\n",
    "        # We perform a second query to pull the label which is just saved as is\n",
    "        x = np.float32(np.array([float(pixel) for pixel in self.db.execute(\"SELECT pixels from \"+self.keyspace+\".\"+self.table+\" WHERE id = \"+str(index)+\";\").one()[0]]).reshape(28,28)/255)\n",
    "        y = self.db.execute(\"SELECT label from \"+self.keyspace+\".\"+self.table+\" WHERE id = \"+str(index)+\";\").one()[0]\n",
    "        if self.transform:\n",
    "            x = self.transform(x)\n",
    "        return x, y\n",
    "    # The len function just returns the length value specified at creation, the Dataset can cause errors if there is not enough rows in the table to furnish the entire lengths worth of data\n",
    "    def __len__(self):\n",
    "        return self.length\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# More imports, this time for the acutal model definition and training\n",
    "# From Pytorch importing the Neural Net type for model defintions, the mathmatical funcions for it as F and the optimization functions\n",
    "import torch.nn as nn\n",
    "import torch.nn.functional as F\n",
    "import torch.optim as optim\n",
    "# Import pickle (python built-in module) for model serialization and storage\n",
    "import pickle as pkl\n",
    "#Auth is custom package containing the actual details for connecting to the AstraDB\n",
    "import auth"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Using auth module to create the cloud_config and auth_provider objects that the loaders need and creating a session so we can submit queries manually as well\n",
    "cloud_config = {'secure_connect_bundle': auth.scb_path}\n",
    "auth_provider = PlainTextAuthProvider(auth.auth_id, auth.auth_token)\n",
    "cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider)\n",
    "session = cluster.connect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating our datasets for the train set and test set, and their associated loaders, which pytorch will use to feed in data for the trianing process.\n",
    "#The transforms we provide here are a composition of the toTensor tranform which takes the numpy array of data that we have and turns it into a pytorch tensor object\n",
    "# and the Nornalize .1307,.3081 tranform, which ensures that the data has 0 mean and unit standard deviation\n",
    "train_dataset = AstraDataset(   \n",
    "                    cloud_config, \n",
    "                    auth_provider, \n",
    "                    \"mnist_digits\", \n",
    "                    \"raw_train\", \n",
    "                    100, \n",
    "                    transforms.Compose([\n",
    "                               transforms.ToTensor(),\n",
    "                               transforms.Normalize((0.1307,), (0.3081,))])\n",
    "                             )\n",
    "train_loader = DataLoader(train_dataset, batch_size=10, shuffle=True)\n",
    "\n",
    "test_dataset = AstraDataset(\n",
    "                    cloud_config, \n",
    "                    auth_provider, \n",
    "                    \"mnist_digits\", \n",
    "                    \"raw_test\", \n",
    "                    100, \n",
    "                    transforms.Compose([\n",
    "                               transforms.ToTensor(),\n",
    "                               transforms.Normalize((0.1307,), (0.3081,))])\n",
    "                             )\n",
    "test_loader = DataLoader(test_dataset, batch_size=10, shuffle=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Constant associated with the training of our neural net\n",
    "# Number of epochs to train the model for\n",
    "n_epochs = 1\n",
    "# How many examples to use for training during a single epoch\n",
    "batch_size_train = 64\n",
    "#How many examples to use for testing during a single epoch\n",
    "batch_size_test = 1000\n",
    "# How big of a change a single backprogatation step is allowed to make to the model\n",
    "learning_rate = 0.01\n",
    "# How much the change in model weights gets carried between backprogatiton steps\n",
    "momentum = 0.5\n",
    "# Logging settings\n",
    "log_interval = 10\n",
    "\n",
    "# Staring seed value, change this to get a different outcome from even a single epoch of training\n",
    "random_seed = 1\n",
    "torch.backends.cudnn.enabled = False\n",
    "torch.manual_seed(random_seed)\n",
    "\n",
    "# Pull one set of example data from the loader\n",
    "examples = enumerate(test_loader)\n",
    "# Extract one example from that set\n",
    "batch_idx, (example_data, example_targets) = next(examples)\n",
    "# Show the shape of the training example\n",
    "print(example_data.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Defining the structure and forward propagation steps of our neural network\n",
    "class Net(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(Net, self).__init__()\n",
    "        # Defining the shapes of the individual layers used. Here we define two convolutional layers (specialized for image processing), a dropout layer, and two linear layers.\n",
    "        self.conv1 = nn.Conv2d(1, 10, kernel_size=5)\n",
    "        self.conv2 = nn.Conv2d(10, 20, kernel_size=5)\n",
    "        self.conv2_drop = nn.Dropout2d()\n",
    "        self.fc1 = nn.Linear(320, 50)\n",
    "        self.fc2 = nn.Linear(50, 10)\n",
    "    # Defining how those layers are connected and what functions are being used internal to the neruons. First the first convolutional layer, then the second followed by the dropout layer,\n",
    "    # followed by linear 1 and then linear 2.\n",
    "    def forward(self, x):\n",
    "        x = F.relu(F.max_pool2d(self.conv1(x), 2))\n",
    "        x = F.relu(F.max_pool2d(self.conv2_drop(self.conv2(x)), 2))\n",
    "        x = x.view(-1, 320)\n",
    "        x = F.relu(self.fc1(x))\n",
    "        x = F.dropout(x, training=self.training)\n",
    "        x = self.fc2(x)\n",
    "        return F.log_softmax(x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create our network object and the Optimizer that has the backprogatation setting included.\n",
    "network = Net()\n",
    "optimizer = optim.SGD(network.parameters(), lr=learning_rate,\n",
    "                      momentum=momentum)\n",
    "# These lists store tje loss value functions over the training epoch\n",
    "train_losses = []\n",
    "train_counter = []\n",
    "test_losses = []\n",
    "test_counter = [i*len(train_loader.dataset) for i in range(n_epochs + 1)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defines our training function for a single epoch\n",
    "# For everyu piece of data in the trianing loader we calculate the loss function and then do a back propagation step, before outputting our trianing loss and saving the model to Astra\n",
    "def train(epoch):\n",
    "    network.train()\n",
    "    for batch_idx, (data, target) in enumerate(train_loader):\n",
    "        optimizer.zero_grad()\n",
    "        output = network(data)\n",
    "        loss = F.nll_loss(output, target)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        if batch_idx % log_interval == 0:\n",
    "            description_string = 'Train Epoch: {} [{}/{} ({:.0f}%)]\\tLoss: {:.6f}'.format(\n",
    "                epoch, batch_idx * len(data), len(train_loader.dataset),\n",
    "                100. * batch_idx / len(train_loader), loss.item())\n",
    "            train_losses.append(loss.item())\n",
    "            train_counter.append(\n",
    "                (batch_idx*64) + ((epoch-1)*len(train_loader.dataset)))\n",
    "            # Here we save the model to Astra. We take the network state and optimizer state dictionaries and pickle them, dumop them to a bytes object which is cast to a bytearray which\n",
    "            # the cassandra connected can understand. Then we insert them along with the time, a uuid, and the loss function description into Astra.\n",
    "            network_state = bytearray(pkl.dumps(network.state_dict()))\n",
    "            optimizer_state = bytearray(pkl.dumps(optimizer.state_dict()))\n",
    "            print(type(network_state))\n",
    "            #Original save functions used torch.save to save data into .pth files on the disk\n",
    "            #torch.save(network.state_dict(), 'results/model.pth')\n",
    "            #torch.save(optimizer.state_dict(), 'results/optimizer.pth')\n",
    "            query = \"INSERT INTO mnist_digits.models (id, network, optimizer, upload_date, comments) VALUES (uuid(), %s, %s, toTimestamp(now()), %s);\"\n",
    "            values = [network_state, optimizer_state, description_string]\n",
    "            session.execute(query, values)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Defines the test function which takes a network and tests it on the data in the test loader, and returns a loss and accuracy just like during training, but on different data.\n",
    "def test():\n",
    "    network.eval()\n",
    "    test_loss = 0\n",
    "    correct = 0\n",
    "    with torch.no_grad():\n",
    "        for data, target in test_loader:\n",
    "            output = network(data)\n",
    "            test_loss += F.nll_loss(output, target, size_average=False).item()\n",
    "            pred = output.data.max(1, keepdim=True)[1]\n",
    "            correct += pred.eq(target.data.view_as(pred)).sum()\n",
    "    test_loss /= len(test_loader.dataset)\n",
    "    test_losses.append(test_loss)\n",
    "    print('\\nTest set: Avg. loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\\n'.format(\n",
    "        test_loss, correct, len(test_loader.dataset),\n",
    "        100. * correct / len(test_loader.dataset)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#First we call test for our uninitialized model (has random model weight), then for however many epochs were specified we train for an epoch and then test again\n",
    "test()\n",
    "for epoch in range(1, n_epochs + 1):\n",
    "  train(epoch)\n",
    "  test()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.1"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}