{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "title: \"Keras/TensorFlow Example - MNIST\"\n", "date: 2021-02-24\n", "type: technical_note\n", "draft: false\n", "---" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Tensorflow Keras example with SavedModel model saving\n", "---\n", "\n", "

Tested with TensorFlow 2.4.0

" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "

\n", "

Machine Learning on Hopsworks\n", "

\n", "

\n", "\n", "![hops.png](../../images/hops.png)\n", "\n", "## The `hops` python module\n", "\n", "`hops` is a helper library for Hops that facilitates development by hiding the complexity of running applications and iteracting with services.\n", "\n", "Have a feature request or encountered an issue? Please let us know on github.\n", "\n", "### Using the `experiment` module\n", "\n", "To be able to run your Machine Learning code in Hopsworks, the code for the whole program needs to be provided and put inside a wrapper function. Everything, from importing libraries to reading data and defining the model and running the program needs to be put inside a wrapper function.\n", "\n", "The `experiment` module provides an api to Python programs such as TensorFlow, Keras and PyTorch on a Hopsworks on any number of machines and GPUs.\n", "\n", "An Experiment could be a single Python program, which we refer to as an **Experiment**. \n", "\n", "Grid search or genetic hyperparameter optimization such as differential evolution which runs several Experiments in parallel, which we refer to as **Parallel Experiment**. \n", "\n", "ParameterServerStrategy, CollectiveAllReduceStrategy and MultiworkerMirroredStrategy making multi-machine/multi-gpu training as simple as invoking a function for orchestration. This mode is referred to as **Distributed Training**.\n", "\n", "### Using the `tensorboard` module\n", "The `tensorboard` module allow us to get the log directory for summaries and checkpoints to be written to the TensorBoard we will see in a bit. The only function that we currently need to call is `tensorboard.logdir()`, which returns the path to the TensorBoard log directory. Furthermore, the content of this directory will be put in as a Dataset in your project's Experiments folder.\n", "\n", "The directory could in practice be used to store other data that should be accessible after the experiment is finished.\n", "```python\n", "# Use this module to get the TensorBoard logdir\n", "from hops import tensorboard\n", "tensorboard_logdir = tensorboard.logdir()\n", "```\n", "\n", "### Using the `hdfs` module\n", "The `hdfs` module provides a method to get the path in HopsFS where your data is stored, namely by calling `hdfs.project_path()`. The path resolves to the root path for your project, which is the view that you see when you click `Data Sets` in HopsWorks. To point where your actual data resides in the project you to append the full path from there to your Dataset. For example if you create a mnist folder in your Resources Dataset, the path to the mnist data would be `hdfs.project_path() + 'Resources/mnist'`\n", "\n", "```python\n", "# Use this module to get the path to your project in HopsFS, then append the path to your Dataset in your project\n", "from hops import hdfs\n", "project_path = hdfs.project_path()\n", "```\n", "\n", "```python\n", "# Downloading the mnist dataset to the current working directory\n", "from hops import hdfs\n", "mnist_hdfs_path = hdfs.project_path() + \"Resources/mnist\"\n", "local_mnist_path = hdfs.copy_to_local(mnist_hdfs_path)\n", "```\n", "\n", "### Documentation\n", "See the following links to learn more about running experiments in Hopsworks\n", "\n", "- Learn more about experiments\n", "
\n", "- Building End-To-End pipelines\n", "
\n", "- Give us a star, create an issue or a feature request on Hopsworks github\n", "\n", "### Managing experiments\n", "Experiments service provides a unified view of all the experiments run using the `experiment` module.\n", "
\n", "As demonstrated in the gif it provides general information about the experiment and the resulting metric. Experiments can be visualized meanwhile or after training in a TensorBoard.\n", "
\n", "
\n", "![Image7-Monitor.png](../../images/experiments.gif)" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [], "source": [ "def keras_mnist():\n", " \n", " import os\n", " import sys\n", " import uuid\n", " import random\n", " \n", " import numpy as np\n", " \n", " from tensorflow import keras\n", " import tensorflow as tf\n", " from tensorflow.keras.models import Sequential\n", " from tensorflow.keras.layers import Dense, Dropout, Flatten\n", " from tensorflow.keras.layers import Conv2D, MaxPooling2D\n", " from tensorflow.keras.callbacks import TensorBoard\n", " from tensorflow.keras import backend as K\n", "\n", " import math\n", " from hops import tensorboard\n", "\n", " from hops import model as hops_model\n", " from hops import hdfs\n", "\n", " batch_size=32\n", " \n", " num_classes = 10\n", " \n", " # Provide path to train and validation datasets\n", " train_filenames = [hdfs.project_path() + \"TourData/mnist/train/train.tfrecords\"]\n", " validation_filenames = [hdfs.project_path() + \"TourData/mnist/validation/validation.tfrecords\"]\n", " \n", " # Define input function\n", " def data_input(filenames, batch_size=32, num_classes = 10, shuffle=False, repeat=None):\n", "\n", " def parser(serialized_example):\n", " \"\"\"Parses a single tf.Example into image and label tensors.\"\"\"\n", " features = tf.io.parse_single_example(\n", " serialized_example,\n", " features={\n", " 'image_raw': tf.io.FixedLenFeature([], tf.string),\n", " 'label': tf.io.FixedLenFeature([], tf.int64),\n", " })\n", " image = tf.io.decode_raw(features['image_raw'], tf.uint8)\n", " image.set_shape([28 * 28])\n", "\n", " # Normalize the values of the image from the range [0, 255] to [-0.5, 0.5]\n", " image = tf.cast(image, tf.float32) / 255 - 0.5\n", " label = tf.cast(features['label'], tf.int32)\n", " \n", " # Create a one hot array for your labels\n", " label = tf.one_hot(label, num_classes)\n", " \n", " return image, label\n", "\n", " # Import MNIST data\n", " dataset = tf.data.TFRecordDataset(filenames)\n", "\n", " # Map the parser over dataset, and batch results by up to batch_size\n", " dataset = dataset.map(parser)\n", " if shuffle:\n", " dataset = dataset.shuffle(buffer_size=128)\n", " dataset = dataset.batch(batch_size, drop_remainder=True)\n", " dataset = dataset.repeat(repeat)\n", " return dataset\n", "\n", " # Define a Keras Model.\n", " model = tf.keras.Sequential()\n", " model.add(tf.keras.layers.Dense(128, activation='relu', input_shape=(784,)))\n", " model.add(tf.keras.layers.Dense(num_classes, activation='softmax'))\n", "\n", " # Compile the model.\n", " model.compile(loss=tf.keras.losses.categorical_crossentropy,\n", " optimizer= tf.keras.optimizers.Adam(0.001),\n", " metrics=['accuracy']\n", " )\n", " \n", " callbacks = [\n", " tf.keras.callbacks.TensorBoard(log_dir=tensorboard.logdir()),\n", " tf.keras.callbacks.ModelCheckpoint(filepath=tensorboard.logdir()),\n", " ]\n", " model.fit(data_input(train_filenames, batch_size), \n", " verbose=0,\n", " epochs=3, \n", " steps_per_epoch=5,\n", " validation_data=data_input(validation_filenames, batch_size),\n", " validation_steps=1, \n", " callbacks=callbacks\n", " )\n", " \n", " score = model.evaluate(data_input(validation_filenames, batch_size), steps=1)\n", "\n", " # Export model\n", " # WARNING(break-tutorial-inline-code): The following code snippet is\n", " # in-lined in tutorials, please update tutorial documents accordingly\n", " # whenever code changes.\n", "\n", " export_path = os.getcwd() + '/model-' + str(uuid.uuid4())\n", " print('Exporting trained model to: {}'.format(export_path))\n", " \n", " tf.saved_model.save(model, export_path)\n", "\n", " print('Done exporting!')\n", " \n", " metrics = {'accuracy': score[1]}\n", " \n", " hops_model.export(export_path, \"mnist\", metrics=metrics) \n", " \n", " return metrics" ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Finished Experiment \n", "\n", "('hdfs://rpc.namenode.service.consul:8020/Projects/demo_ml_meb10000/Experiments/application_1621614908761_0013_7', {'accuracy': 0.78125, 'log': 'Experiments/application_1621614908761_0013_7/output.log'})" ] } ], "source": [ "from hops import experiment\n", "from hops import hdfs\n", "\n", "experiment.launch(keras_mnist, name='keras mnist', local_logdir=True, metric_key='accuracy')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "PySpark", "language": "python", "name": "pysparkkernel" }, "language_info": { "codemirror_mode": { "name": "python", "version": 3 }, "mimetype": "text/x-python", "name": "pyspark", "pygments_lexer": "python3" } }, "nbformat": 4, "nbformat_minor": 4 }