hyperparameter-opt-spark-tf.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>0</td><td>application_1513605290137_0001</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://hopsworks0:8088/proxy/application_1513605290137_0001/\">Link</a></td><td><a target=\"_blank\" href=\"http://hopsworks0:8042/node/containerlogs/container_e01_1513605290137_0001_01_000001/demo_tensorflow_admin000__meb10000\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "def launch(spark_session, map_fun, args_dict):\n",
    "    \"\"\" Run the wrapper function with each hyperparameter combination as specified by the dictionary\n",
    "\n",
    "    Args:\n",
    "      :spark_session: SparkSession object\n",
    "      :map_fun: The TensorFlow function to run\n",
    "      :args_dict: A dictionary containing hyperparameter values to insert as arguments for each TensorFlow job\n",
    "    \"\"\"\n",
    "\n",
    "    sc = spark_session.sparkContext\n",
    "\n",
    "    # Length of the list of the first list of arguments represents the number of Spark tasks\n",
    "    num_tasks = len(args_dict.values()[0])\n",
    "\n",
    "    # Create a number of partitions (tasks)\n",
    "    nodeRDD = sc.parallelize(range(num_tasks), num_tasks)\n",
    "\n",
    "    # Execute each of the hyperparameter arguments as a task\n",
    "    nodeRDD.foreachPartition(_do_search(map_fun, args_dict))\n",
    "\n",
    "\n",
    "def _do_search(map_fun, args_dict):\n",
    "    \n",
    "    def _wrapper_fun(iter):\n",
    "\n",
    "        for i in iter:\n",
    "            executor_num = i\n",
    "\n",
    "        argcount = map_fun.func_code.co_argcount\n",
    "        names = map_fun.func_code.co_varnames\n",
    "\n",
    "        args = []\n",
    "        argIndex = 0\n",
    "        while argcount > 0:\n",
    "            # Get arguments for hyperparameter combination\n",
    "            param_name = names[argIndex]\n",
    "            param_val = args_dict[param_name][executor_num]\n",
    "            args.append(param_val)\n",
    "            argcount -= 1\n",
    "            argIndex += 1\n",
    "        map_fun(*args)\n",
    "    return _wrapper_fun"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mnist(num_steps):\n",
    "    from tensorflow.examples.tutorials.mnist import input_data\n",
    "    import tensorflow as tf\n",
    "    \n",
    "    mnist = input_data.read_data_sets('/tmp/tensorflow/mnist/input_data', one_hot=True)\n",
    "    # Create the model\n",
    "    x = tf.placeholder(tf.float32, [None, 784])\n",
    "    W = tf.Variable(tf.zeros([784, 10]))\n",
    "    b = tf.Variable(tf.zeros([10]))\n",
    "    y = tf.matmul(x, W) + b\n",
    "    \n",
    "    # Define loss and optimizer\n",
    "    y_ = tf.placeholder(tf.float32, [None, 10])\n",
    "    \n",
    "    cross_entropy = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(labels=y_, logits=y))\n",
    "    train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)\n",
    "    \n",
    "    sess = tf.InteractiveSession()\n",
    "    tf.global_variables_initializer().run()\n",
    "    for _ in range(num_steps):\n",
    "        batch_xs, batch_ys = mnist.train.next_batch(100)\n",
    "        sess.run(train_step, feed_dict={x: batch_xs, y_: batch_ys})\n",
    "        \n",
    "    # Test trained model\n",
    "    correct_prediction = tf.equal(tf.argmax(y, 1), tf.argmax(y_, 1))\n",
    "    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))\n",
    "    print(sess.run(accuracy, feed_dict={x: mnist.test.images, y_: mnist.test.labels}))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "args_dict = {'num_steps': [1000, 10000]}\n",
    "launch(spark, mnist, args_dict)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}