{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# MultiWorkerMirroredStrategy on Hopsworks\n", "---\n", "\n", "

Tested with TensorFlow 2.3.0

\n", "\n", "

\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": 1, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Starting Spark application\n" ] }, { "data": { "text/html": [ "\n", "
IDYARN Application IDKindStateSpark UIDriver log
24application_1600264891477_0030pysparkidleLinkLink
" ], "text/plain": [ "" ] }, "metadata": {}, "output_type": "display_data" }, { "name": "stdout", "output_type": "stream", "text": [ "SparkSession available as 'spark'.\n" ] } ], "source": [ "def multi_worker_mirrored_training():\n", " import sys\n", " import numpy as np\n", " import tensorflow as tf\n", " from hops import tensorboard\n", " from hops import devices\n", " model_dir = tensorboard.logdir()\n", " print('Using %s to store checkpoints.' % model_dir)\n", " # Define distribution strategy\n", " strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()\n", " # Define per device batch size\n", " batch_size_per_replica = 32\n", " # Define global batch size \n", " batch_size = batch_size_per_replica * strategy.num_replicas_in_sync\n", " # Define model hyper parameters\n", " epochs=30 \n", " steps_per_epoch=5\n", " validation_steps=2 \n", " shuffle_size = batch_size * 4\n", " def input_fn():\n", " x = np.random.random((1024, 10))\n", " y = np.random.randint(2, size=(1024, 1))\n", " x = tf.cast(x, tf.float32)\n", " dataset = tf.data.Dataset.from_tensor_slices((x, y))\n", " dataset = dataset.repeat(500)\n", " dataset = dataset.batch(batch_size)\n", " options = tf.data.Options()\n", " options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.OFF\n", " return dataset.with_options(options)\n", " with strategy.scope():\n", " # Define a Keras Model.\n", " model = tf.keras.Sequential()\n", " model.add(tf.keras.layers.Dense(16, activation='relu', input_shape=(10,)))\n", " model.add(tf.keras.layers.Dense(1, activation='sigmoid'))\n", " # Compile the model.\n", " model.compile(loss=tf.keras.losses.BinaryCrossentropy(),\n", " optimizer=tf.keras.optimizers.Adam(),\n", " metrics=['accuracy']\n", " )\n", " callbacks = [\n", " tf.keras.callbacks.TensorBoard(log_dir=model_dir),\n", " tf.keras.callbacks.ModelCheckpoint(filepath=model_dir),\n", " ]\n", " model.fit(input_fn(), \n", " verbose=0, \n", " epochs=epochs, \n", " steps_per_epoch=steps_per_epoch,\n", " validation_data=input_fn(),\n", " validation_steps=validation_steps, \n", " callbacks=callbacks\n", " )\n", " score = model.evaluate(input_fn(), steps=1)\n", " metrics = {'accuracy': score[1]}\n", " return metrics " ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Finished Experiment \n", "\n", "('hdfs://rpc.namenode.service.consul:8020/Projects/demo_deep_learning_admin000/Experiments/application_1600264891477_0030_1', {'accuracy': 0.4375, 'log': 'Experiments/application_1600264891477_0030_1/chief_0_output.log'})" ] } ], "source": [ "from hops import experiment\n", "experiment.mirrored(multi_worker_mirrored_training, name='random data model', metric_key='accuracy')" ] } ], "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 }