mnist.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# MNIST using Distributed Keras\n",
    "\n",
    "**Joeri Hermans** (Technical Student, IT-DB-SAS, CERN)             \n",
    "*Departement of Knowledge Engineering*         \n",
    "*Maastricht University, The Netherlands*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!(date +%d\\ %B\\ %G)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook we will show you how to process the [MNIST](http://yann.lecun.com/exdb/mnist/) dataset using Distributed Keras. As in the [workflow](https://github.com/JoeriHermans/dist-keras/blob/master/examples/workflow.ipynb) notebook, we will guide you through the complete machine learning pipeline.\n",
    "\n",
    "## Preparation\n",
    "\n",
    "To get started, we first load all the required imports. Please make sure you installed `dist-keras`, and `seaborn`. Furthermore, we assume that you have access to an installation which provides Apache Spark.\n",
    "\n",
    "Before you start this notebook, place the MNIST dataset (which is provided in this repository) on HDFS. Or in the case HDFS is not available, place it on the local filesystem. But make sure the path to the file is identical for all computing nodes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "\n",
    "import numpy as np\n",
    "\n",
    "import seaborn as sns\n",
    "\n",
    "from keras.optimizers import *\n",
    "from keras.models import Sequential\n",
    "from keras.layers.core import *\n",
    "from keras.layers.convolutional import *\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SparkConf\n",
    "\n",
    "from matplotlib import pyplot as plt\n",
    "import matplotlib.patches as mpatches\n",
    "\n",
    "from pyspark.ml.feature import StandardScaler\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.feature import OneHotEncoder\n",
    "from pyspark.ml.feature import MinMaxScaler\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "from distkeras.trainers import *\n",
    "from distkeras.predictors import *\n",
    "from distkeras.transformers import *\n",
    "from distkeras.evaluators import *\n",
    "from distkeras.utils import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the following cell, adapt the parameters to fit your personal requirements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Modify these variables according to your needs.\n",
    "application_name = \"Distributed Keras MNIST Notebook\"\n",
    "using_spark_2 = False\n",
    "local = False\n",
    "path_train = \"data/mnist_train.csv\"\n",
    "path_test = \"data/mnist_test.csv\"\n",
    "if local:\n",
    "    # Tell master to use local resources.\n",
    "    master = \"local[*]\"\n",
    "    num_processes = 3\n",
    "    num_executors = 1\n",
    "else:\n",
    "    # Tell master to use YARN.\n",
    "    master = \"yarn-client\"\n",
    "    num_executors = 20\n",
    "    num_processes = 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# This variable is derived from the number of cores and executors, and will be used to assign the number of model trainers.\n",
    "num_workers = num_executors * num_processes\n",
    "\n",
    "print(\"Number of desired executors: \" + `num_executors`)\n",
    "print(\"Number of desired processes / executor: \" + `num_processes`)\n",
    "print(\"Total number of workers: \" + `num_workers`)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "# Use the DataBricks CSV reader, this has some nice functionality regarding invalid values.\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages com.databricks:spark-csv_2.10:1.4.0 pyspark-shell'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "conf = SparkConf()\n",
    "conf.set(\"spark.app.name\", application_name)\n",
    "conf.set(\"spark.master\", master)\n",
    "conf.set(\"spark.executor.cores\", `num_processes`)\n",
    "conf.set(\"spark.executor.instances\", `num_executors`)\n",
    "conf.set(\"spark.executor.memory\", \"4g\")\n",
    "conf.set(\"spark.locality.wait\", \"0\")\n",
    "conf.set(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\");\n",
    "\n",
    "# Check if the user is running Spark 2.0 +\n",
    "if using_spark_2:\n",
    "    sc = SparkSession.builder.config(conf=conf) \\\n",
    "            .appName(application_name) \\\n",
    "            .getOrCreate()\n",
    "else:\n",
    "    # Create the Spark context.\n",
    "    sc = SparkContext(conf=conf)\n",
    "    # Add the missing imports\n",
    "    from pyspark import SQLContext\n",
    "    sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Check if we are using Spark 2.0\n",
    "if using_spark_2:\n",
    "    reader = sc\n",
    "else:\n",
    "    reader = sqlContext\n",
    "# Read the training dataset.\n",
    "raw_dataset_train = reader.read.format('com.databricks.spark.csv') \\\n",
    "                          .options(header='true', inferSchema='true') \\\n",
    "                          .load(path_train)\n",
    "# Read the testing dataset.\n",
    "raw_dataset_test = reader.read.format('com.databricks.spark.csv') \\\n",
    "                         .options(header='true', inferSchema='true') \\\n",
    "                         .load(path_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As shown in the output of the cell above, we see that every pixel is associated with a seperate column. In order to ensure compatibility with Apache Spark, we vectorize the columns, and add the resulting vectors as a seperate column. However, in order to achieve this, we first need a list of the required columns. This is shown in the cell below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# First, we would like to extract the desired features from the raw dataset.\n",
    "# We do this by constructing a list with all desired columns.\n",
    "# This is identical for the test set.\n",
    "features = raw_dataset_train.columns\n",
    "features.remove('label')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once we have a list of columns names, we can pass this to Spark's [VectorAssembler](http://spark.apache.org/docs/latest/ml-features.html#vectorassembler). This VectorAssembler will take a list of features, vectorize them, and place them in a column defined in `outputCol`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Next, we use Spark's VectorAssembler to \"assemble\" (create) a vector of all desired features.\n",
    "# http://spark.apache.org/docs/latest/ml-features.html#vectorassembler\n",
    "vector_assembler = VectorAssembler(inputCols=features, outputCol=\"features\")\n",
    "# This transformer will take all columns specified in features, and create an additional column \"features\" which will contain all the desired features aggregated into a single vector.\n",
    "dataset_train = vector_assembler.transform(raw_dataset_train)\n",
    "dataset_test = vector_assembler.transform(raw_dataset_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once we have the inputs for our Neural Network (features column) after applying the VectorAssembler, we should also define the outputs. Since we are dealing with a classification task, the output of our Neural Network should be a one-hot encoded vector with 10 elements. For this, we provide a `OneHotTransformer` which accomplish this exact task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Define the number of output classes.\n",
    "nb_classes = 10\n",
    "encoder = OneHotTransformer(nb_classes, input_col=\"label\", output_col=\"label_encoded\")\n",
    "dataset_train = encoder.transform(dataset_train)\n",
    "dataset_test = encoder.transform(dataset_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MNIST\n",
    "\n",
    "[MNIST](http://yann.lecun.com/exdb/mnist/) is a dataset of handwritten digits. Every image is a 28 by 28 pixel grayscale image. This means that every pixel has a value between 0 and 255. Some examples of instances within this dataset are shown in the cells below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def show_instances(column):\n",
    "    global dataset\n",
    "\n",
    "    num_instances = 6 # Number of instances you would like to draw.\n",
    "    x_dimension   = 3 # Number of images to draw on the x-axis.\n",
    "    y_dimension   = 2 # Number of images to draw on the y-axis.\n",
    "\n",
    "    # Fetch 3 different instance from the dataset.\n",
    "    instances = dataset_train.select(column).take(num_instances)\n",
    "    # Process the instances.\n",
    "    for i in range(0, num_instances):\n",
    "        instance = instances[i]\n",
    "        instance = instance[column].toArray().reshape((28, 28))\n",
    "        instances[i] = instance\n",
    "\n",
    "    # Draw the sampled instances.\n",
    "    fig, axn = plt.subplots(y_dimension, x_dimension, sharex=True, sharey=True)\n",
    "    num_axn = len(axn.flat)\n",
    "    for i in range(0, num_axn):\n",
    "        ax = axn.flat[i]\n",
    "        h = sns.heatmap(instances[i], ax=ax)\n",
    "        h.set_yticks([])\n",
    "        h.set_xticks([])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "show_instances(\"features\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Normalization\n",
    "\n",
    "In this Section, we will normalize the feature vectors between the 0 and 1 range."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Clear the dataset in the case you ran this cell before.\n",
    "dataset_train = dataset_train.select(\"features\", \"label\", \"label_encoded\")\n",
    "dataset_test = dataset_test.select(\"features\", \"label\", \"label_encoded\")\n",
    "# Allocate a MinMaxTransformer using Distributed Keras.\n",
    "# o_min -> original_minimum\n",
    "# n_min -> new_minimum\n",
    "transformer = MinMaxTransformer(n_min=0.0, n_max=1.0, \\\n",
    "                                o_min=0.0, o_max=250.0, \\\n",
    "                                input_col=\"features\", \\\n",
    "                                output_col=\"features_normalized\")\n",
    "# Transform the dataset.\n",
    "dataset_train = transformer.transform(dataset_train)\n",
    "dataset_test = transformer.transform(dataset_test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "show_instances(\"features_normalized\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convolutions\n",
    "\n",
    "In order to make the dense vectors compatible with convolution operations in Keras, we add another column which contains the matrix form of these images. We provide a utility class (MatrixTransformer), which helps you with this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "reshape_transformer = ReshapeTransformer(\"features_normalized\", \"matrix\", (28, 28, 1))\n",
    "dataset_train = reshape_transformer.transform(dataset_train)\n",
    "dataset_test = reshape_transformer.transform(dataset_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Development"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Multilayer Perceptron"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "mlp = Sequential()\n",
    "mlp.add(Dense(1000, input_shape=(784,)))\n",
    "mlp.add(Activation('relu'))\n",
    "mlp.add(Dense(250))\n",
    "mlp.add(Activation('relu'))\n",
    "mlp.add(Dense(10))\n",
    "mlp.add(Activation('softmax'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "mlp.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "optimizer_mlp = 'adam'\n",
    "loss_mlp = 'categorical_crossentropy'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Convolutional network"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Taken from Keras MNIST example.\n",
    "\n",
    "# Declare model parameters.\n",
    "img_rows, img_cols = 28, 28\n",
    "# number of convolutional filters to use\n",
    "nb_filters = 32\n",
    "# size of pooling area for max pooling\n",
    "pool_size = (2, 2)\n",
    "# convolution kernel size\n",
    "kernel_size = (3, 3)\n",
    "input_shape = (img_rows, img_cols, 1)\n",
    "\n",
    "# Construct the model.\n",
    "convnet = Sequential()\n",
    "convnet.add(Convolution2D(nb_filters, kernel_size[0], kernel_size[1],\n",
    "                          border_mode='valid',\n",
    "                          input_shape=input_shape))\n",
    "convnet.add(Activation('relu'))\n",
    "convnet.add(Convolution2D(nb_filters, kernel_size[0], kernel_size[1]))\n",
    "convnet.add(Activation('relu'))\n",
    "convnet.add(MaxPooling2D(pool_size=pool_size))\n",
    "\n",
    "convnet.add(Flatten())\n",
    "convnet.add(Dense(225))\n",
    "convnet.add(Activation('relu'))\n",
    "convnet.add(Dense(nb_classes))\n",
    "convnet.add(Activation('softmax'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "convnet.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "optimizer_convnet = 'adam'\n",
    "loss_convnet = 'categorical_crossentropy'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluation\n",
    "\n",
    "We define a utility function which will compute the accuracy for us."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def evaluate_accuracy(model, test_set, features=\"features_normalized_dense\"):\n",
    "    evaluator = AccuracyEvaluator(prediction_col=\"prediction_index\", label_col=\"label\")\n",
    "    predictor = ModelPredictor(keras_model=model, features_col=features)\n",
    "    transformer = LabelIndexTransformer(output_dim=nb_classes)\n",
    "    test_set = test_set.select(features, \"label\")\n",
    "    test_set = predictor.predict(test_set)\n",
    "    test_set = transformer.transform(test_set)\n",
    "    score = evaluator.evaluate(test_set)\n",
    "    \n",
    "    return score"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dataset_train.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "dataset_train = dataset_train.select(\"features_normalized\", \"matrix\",\"label\", \"label_encoded\")\n",
    "dataset_test = dataset_test.select(\"features_normalized\", \"matrix\",\"label\", \"label_encoded\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dense_transformer = DenseTransformer(input_col=\"features_normalized\", output_col=\"features_normalized_dense\")\n",
    "dataset_train = dense_transformer.transform(dataset_train)\n",
    "dataset_test = dense_transformer.transform(dataset_test)\n",
    "dataset_train.repartition(num_workers)\n",
    "dataset_test.repartition(num_workers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Assing the training and test set.\n",
    "training_set = dataset_train.repartition(num_workers)\n",
    "test_set = dataset_test.repartition(num_workers)\n",
    "# Cache them.\n",
    "training_set.cache()\n",
    "test_set.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(training_set.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### DOWNPOUR (Multilayer Perceptron)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer = DOWNPOUR(keras_model=mlp, worker_optimizer=optimizer_mlp, loss=loss_mlp, num_workers=num_workers,\n",
    "                   batch_size=4, communication_window=5, num_epoch=1,\n",
    "                   features_col=\"features_normalized_dense\", label_col=\"label_encoded\")\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ADAG (MultiLayer Perceptron)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer = ADAG(keras_model=mlp, worker_optimizer=optimizer_mlp, loss=loss_mlp, num_workers=num_workers,\n",
    "               batch_size=4, communication_window=15, num_epoch=1,\n",
    "               features_col=\"features_normalized_dense\", label_col=\"label_encoded\")\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### EASGD (MultiLayer Perceptron)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "trainer = AEASGD(keras_model=mlp, worker_optimizer=optimizer_mlp, loss=loss_mlp, num_workers=num_workers,\n",
    "                 batch_size=4, communication_window=35, num_epoch=1, features_col=\"features_normalized_dense\",\n",
    "                 label_col=\"label_encoded\")\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### DOWNPOUR (Convolutional network)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer = DOWNPOUR(keras_model=convnet, worker_optimizer=optimizer_convnet, loss=loss_convnet,\n",
    "                   num_workers=num_workers, batch_size=4, communication_window=5,\n",
    "                   num_epoch=1, features_col=\"matrix\", label_col=\"label_encoded\")\n",
    "trainer.set_parallelism_factor(1)\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set, \"matrix\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ADAG (Convolutional network)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer = ADAG(keras_model=convnet, worker_optimizer=optimizer_convnet, loss=loss_convnet,\n",
    "               num_workers=num_workers, batch_size=15, communication_window=5, num_epoch=1,\n",
    "               features_col=\"matrix\", label_col=\"label_encoded\")\n",
    "trainer.set_parallelism_factor(1)\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set, \"matrix\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### EASGD (Convolutional network)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer = AEASGD(keras_model=convnet, worker_optimizer=optimizer_convnet, loss=loss_convnet, \n",
    "                 num_workers=num_workers, batch_size=35, communication_window=32, num_epoch=1,\n",
    "                 features_col=\"matrix\", label_col=\"label_encoded\")\n",
    "trained_model = trainer.train(training_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(\"Training time: \" + str(trainer.get_training_time()))\n",
    "print(\"Accuracy: \" + str(evaluate_accuracy(trained_model, test_set, \"matrix\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainer.parameter_server.num_updates"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}