Code appendix/Exploring distributed training with keras and tensorflow data module.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tensorflow as tf\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = 10\n",
    "batch = 256"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_cnn():\n",
    "    inputs = tf.keras.Input(shape=(28,28,1))\n",
    "    x = tf.keras.layers.Conv2D(8,(3,3),padding='same',activation='relu')(inputs)\n",
    "    x = tf.keras.layers.Flatten()(x)\n",
    "    outputs = tf.keras.layers.Dense(10,activation='softmax')(x)\n",
    "    model = tf.keras.Model(inputs,outputs)\n",
    "    return model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using numpy array as inputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING:tensorflow:From /Users/seanchoi/anaconda3/envs/cv/lib/python3.6/site-packages/tensorflow/python/ops/init_ops.py:1251: calling VarianceScaling.__init__ (from tensorflow.python.ops.init_ops) with dtype is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "Call initializer instance with the dtype argument instead of passing it to the constructor\n",
      "Train on 60000 samples, validate on 10000 samples\n",
      "Epoch 1/10\n",
      "60000/60000 - 2s - loss: 0.5067 - acc: 0.8754 - val_loss: 0.2356 - val_acc: 0.9361\n",
      "Epoch 2/10\n",
      "60000/60000 - 2s - loss: 0.2094 - acc: 0.9414 - val_loss: 0.1704 - val_acc: 0.9527\n",
      "Epoch 3/10\n",
      "60000/60000 - 2s - loss: 0.1540 - acc: 0.9575 - val_loss: 0.1364 - val_acc: 0.9624\n",
      "Epoch 4/10\n",
      "60000/60000 - 2s - loss: 0.1221 - acc: 0.9659 - val_loss: 0.1110 - val_acc: 0.9704\n",
      "Epoch 5/10\n",
      "60000/60000 - 2s - loss: 0.1002 - acc: 0.9721 - val_loss: 0.0964 - val_acc: 0.9728\n",
      "Epoch 6/10\n",
      "60000/60000 - 2s - loss: 0.0846 - acc: 0.9767 - val_loss: 0.0845 - val_acc: 0.9756\n",
      "Epoch 7/10\n",
      "60000/60000 - 2s - loss: 0.0742 - acc: 0.9795 - val_loss: 0.0839 - val_acc: 0.9755\n",
      "Epoch 8/10\n",
      "60000/60000 - 2s - loss: 0.0659 - acc: 0.9812 - val_loss: 0.0778 - val_acc: 0.9775\n",
      "Epoch 9/10\n",
      "60000/60000 - 2s - loss: 0.0590 - acc: 0.9836 - val_loss: 0.0728 - val_acc: 0.9775\n",
      "Epoch 10/10\n",
      "60000/60000 - 2s - loss: 0.0542 - acc: 0.9845 - val_loss: 0.0705 - val_acc: 0.9782\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<tensorflow.python.keras.callbacks.History at 0x63460add8>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(X_train_1,y_train_1),(X_test_1,y_test_1) = tf.keras.datasets.mnist.load_data()\n",
    "X_train_1 = X_train_1.astype(np.float32)/255.\n",
    "X_train_1 = np.expand_dims(X_train_1,axis=-1)\n",
    "X_test_1 = X_test_1.astype(np.float32)/255.\n",
    "X_test_1 = np.expand_dims(X_test_1,axis=-1)\n",
    "y_train_1 = tf.keras.utils.to_categorical(y_train_1,10)\n",
    "y_test_1 = tf.keras.utils.to_categorical(y_test_1,10)\n",
    "\n",
    "model_1 = build_cnn()\n",
    "model_1.compile(loss='categorical_crossentropy',optimizer='adam',metrics=['accuracy'])\n",
    "model_1.fit(X_train_1,y_train_1,epochs=epochs,batch_size=batch,validation_data=(X_test_1,y_test_1),verbose=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10000/10000 [==============================] - 0s 25us/sample - loss: 0.0705 - acc: 0.9782\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[0.07052422738065943, 0.9782]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model_1.evaluate(X_test_1,y_test_1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using tf.data.Dataset as inputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocessing(img,label):\n",
    "    img = tf.cast(img,tf.float64)\n",
    "    img = img/255.\n",
    "    img = tf.expand_dims(img,axis=-1)\n",
    "    label = tf.one_hot(label,10,dtype=tf.int32)\n",
    "    return img,label"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch 1/10\n",
      "234/234 - 3s - loss: 0.4867 - acc: 0.8736 - val_loss: 0.2972 - val_acc: 0.9062\n",
      "Epoch 2/10\n",
      "234/234 - 3s - loss: 0.2206 - acc: 0.9376 - val_loss: 0.2048 - val_acc: 0.9406\n",
      "Epoch 3/10\n",
      "234/234 - 3s - loss: 0.1578 - acc: 0.9561 - val_loss: 0.1610 - val_acc: 0.9578\n",
      "Epoch 4/10\n",
      "234/234 - 3s - loss: 0.1216 - acc: 0.9668 - val_loss: 0.1344 - val_acc: 0.9578\n",
      "Epoch 5/10\n",
      "234/234 - 3s - loss: 0.0994 - acc: 0.9729 - val_loss: 0.1206 - val_acc: 0.9609\n",
      "Epoch 6/10\n",
      "234/234 - 3s - loss: 0.0844 - acc: 0.9769 - val_loss: 0.1083 - val_acc: 0.9633\n",
      "Epoch 7/10\n",
      "234/234 - 3s - loss: 0.0748 - acc: 0.9793 - val_loss: 0.1010 - val_acc: 0.9688\n",
      "Epoch 8/10\n",
      "234/234 - 3s - loss: 0.0665 - acc: 0.9813 - val_loss: 0.0940 - val_acc: 0.9711\n",
      "Epoch 9/10\n",
      "234/234 - 3s - loss: 0.0606 - acc: 0.9832 - val_loss: 0.0909 - val_acc: 0.9742\n",
      "Epoch 10/10\n",
      "234/234 - 3s - loss: 0.0547 - acc: 0.9847 - val_loss: 0.0892 - val_acc: 0.9750\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<tensorflow.python.keras.callbacks.History at 0x6348ed2b0>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(X_train_2,y_train_2),(X_test_2,y_test_2) = tf.keras.datasets.mnist.load_data()\n",
    "\n",
    "train_ds = tf.data.Dataset.from_tensor_slices((X_train_2,y_train_2)).map(preprocessing,num_parallel_calls=tf.data.experimental.AUTOTUNE).shuffle(256).batch(batch).repeat(epochs).prefetch(buffer_size=tf.data.experimental.AUTOTUNE)\n",
    "test_ds = tf.data.Dataset.from_tensor_slices((X_test_2,y_test_2)).map(preprocessing,num_parallel_calls=tf.data.experimental.AUTOTUNE).batch(batch).repeat(epochs).prefetch(buffer_size=tf.data.experimental.AUTOTUNE)\n",
    "\n",
    "model_2 = build_cnn()\n",
    "model_2.compile(loss='categorical_crossentropy',optimizer='adam',metrics=['accuracy'])\n",
    "model_2.fit(train_ds,epochs=epochs,steps_per_epoch=len(X_train_2)//batch,validation_data=test_ds,validation_steps=5,verbose=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "400/400 [==============================] - 3s 8ms/step - loss: 0.0688 - acc: 0.9783\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[0.06881496772402898, 0.9783]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model_2.evaluate(test_ds)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## tensorflow distributed training with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[name: \"/device:CPU:0\"\n",
      "device_type: \"CPU\"\n",
      "memory_limit: 268435456\n",
      "locality {\n",
      "}\n",
      "incarnation: 16748366450226833428\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "from tensorflow.python.client import device_lib\n",
    "print(device_lib.list_local_devices())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING:tensorflow:Not all devices in `tf.distribute.Strategy` are visible to TensorFlow.\n",
      "\n",
      "number of replicas in sync: 1\n",
      "global batch: 256\n"
     ]
    }
   ],
   "source": [
    "(X_train_3,y_train_3),(X_test_3,y_test_3) = tf.keras.datasets.mnist.load_data()\n",
    "\n",
    "mirrored = tf.distribute.MirroredStrategy()\n",
    "print('\\nnumber of replicas in sync: {}'.format(mirrored.num_replicas_in_sync))\n",
    "\n",
    "batch_per_replica = batch\n",
    "global_batch = batch_per_replica * mirrored.num_replicas_in_sync\n",
    "print('global batch: {}'.format(global_batch))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "WARNING:tensorflow:Expected a shuffled dataset but input dataset `x` is not shuffled. Please invoke `shuffle()` on input dataset.\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "INFO:tensorflow:Reduce to /replica:0/task:0/device:CPU:0 then broadcast to ('/replica:0/task:0/device:CPU:0',).\n",
      "Epoch 1/10\n",
      "234/234 - 4s - loss: 0.4543 - acc: 0.8797 - val_loss: 0.3074 - val_acc: 0.9055\n",
      "Epoch 2/10\n",
      "234/234 - 3s - loss: 0.2379 - acc: 0.9323 - val_loss: 0.2360 - val_acc: 0.9227\n",
      "Epoch 3/10\n",
      "234/234 - 4s - loss: 0.1794 - acc: 0.9499 - val_loss: 0.1786 - val_acc: 0.9422\n",
      "Epoch 4/10\n",
      "234/234 - 4s - loss: 0.1372 - acc: 0.9621 - val_loss: 0.1481 - val_acc: 0.9516\n",
      "Epoch 5/10\n",
      "234/234 - 4s - loss: 0.1091 - acc: 0.9703 - val_loss: 0.1274 - val_acc: 0.9633\n",
      "Epoch 6/10\n",
      "234/234 - 3s - loss: 0.0916 - acc: 0.9754 - val_loss: 0.1128 - val_acc: 0.9672\n",
      "Epoch 7/10\n",
      "234/234 - 4s - loss: 0.0784 - acc: 0.9788 - val_loss: 0.1044 - val_acc: 0.9656\n",
      "Epoch 8/10\n",
      "234/234 - 3s - loss: 0.0698 - acc: 0.9812 - val_loss: 0.0962 - val_acc: 0.9719\n",
      "Epoch 9/10\n",
      "234/234 - 3s - loss: 0.0627 - acc: 0.9827 - val_loss: 0.0927 - val_acc: 0.9695\n",
      "Epoch 10/10\n",
      "234/234 - 3s - loss: 0.0561 - acc: 0.9844 - val_loss: 0.0859 - val_acc: 0.9727\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<tensorflow.python.keras.callbacks.History at 0x6408c2d30>"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_ds_dist = tf.data.Dataset.from_tensor_slices((X_train_3,y_train_3)).map(preprocessing,num_parallel_calls=tf.data.experimental.AUTOTUNE).shuffle(256).batch(global_batch).repeat(epochs).prefetch(buffer_size=tf.data.experimental.AUTOTUNE)\n",
    "test_ds_dist = tf.data.Dataset.from_tensor_slices((X_test_3,y_test_3)).map(preprocessing,num_parallel_calls=tf.data.experimental.AUTOTUNE).batch(global_batch).repeat(epochs).prefetch(buffer_size=tf.data.experimental.AUTOTUNE)\n",
    "\n",
    "with mirrored.scope():\n",
    "    model_3 = build_cnn()\n",
    "    model_3.compile(loss='categorical_crossentropy',optimizer='adam',metrics=['accuracy'])\n",
    "model_3.fit(train_ds_dist,epochs=epochs,steps_per_epoch=len(X_train_3)//batch,validation_data=test_ds_dist,validation_steps=5,verbose=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "400/400 [==============================] - 3s 8ms/step - loss: 0.0690 - acc: 0.9781\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[0.0689774940838106, 0.9781]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model_3.evaluate(test_ds_dist)"
   ]
  }
 ],
 "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}