diff --git a/examples/hyperopt/README.md b/examples/hyperopt/README.md
index d13d11a0e..f2d43a1bc 100644
--- a/examples/hyperopt/README.md
+++ b/examples/hyperopt/README.md
@@ -9,7 +9,7 @@ Then from the directory `ray/examples/hyperopt/` run the following.
 
 ```
 source ../../setup-env.sh
-python driver.py
+python driver.py # This will take a minute to first download the MNIST dataset.
 ```
 
 Machine learning algorithms often have a number of *hyperparameters* whose
@@ -33,19 +33,19 @@ choose the following hyperparameters:
 - the standard deviation of the distribution from which to initialize the
 network weights
 
-Suppose that we've defined a Python function `train_cnn`, which takes values for
-these hyperparameters as its input, trains a convolutional network using those
-hyperparameters, and returns the accuracy of the trained model on a validation
-set.
+Suppose that we've defined a Python function `train_cnn_and_compute_accuracy`,
+which takes values for these hyperparameters as its input (along with the
+dataset), trains a convolutional network using those hyperparameters, and
+returns the accuracy of the trained model on a validation set.
 
 ```python
-def train_cnn(hyperparameters):
-  # hyperparameters is a dictionary with keys
+def train_cnn_and_compute_accuracy(hyperparameters, train_images, train_labels, validation_images, validation_labels):
+  # Construct a deep network, train it, and return the validation accuracy.
+  # The argument hyperparameters is a dictionary with keys:
   #   - "learning_rate"
   #   - "batch_size"
   #   - "dropout"
   #   - "stddev"
-  # Train a deep network with the above hyperparameters
   return validation_accuracy
 ```
 
@@ -55,63 +55,48 @@ hyperparameters. For example, we can write the following.
 ```python
 def generate_random_params():
   # Randomly choose values for the hyperparameters
-  learning_rate = 10 ** np.random.uniform(-6, 1)
-  batch_size = np.random.randint(30, 100)
+  learning_rate = 10 ** np.random.uniform(-5, 5)
+  batch_size = np.random.randint(1, 100)
   dropout = np.random.uniform(0, 1)
-  stddev = 10 ** np.random.uniform(-3, 1)
+  stddev = 10 ** np.random.uniform(-5, 5)
   return {"learning_rate": learning_rate, "batch_size": batch_size, "dropout": dropout, "stddev": stddev}
 
 results = []
 for _ in range(100):
   randparams = generate_random_params()
-  results.append((randparams, train_cnn(randparams, epochs)))
+  results.append((randparams, train_cnn_and_compute_accuracy(randparams, epochs)))
 ```
 
 Then we can inspect the contents of `results` and see which set of
 hyperparameters worked the best.
 
 Of course, as there are no dependencies between the different invocations of
-`train_cnn`, this computation could easily be parallelized over multiple cores or
-multiple machines. Let's do that now.
+`train_cnn_and_compute_accuracy`, this computation could easily be parallelized
+over multiple cores or multiple machines. Let's do that now.
 
 ### The distributed version
 
-To run this example in Ray, we use three files.
-
-- [driver.py](driver.py) - This is the script that gets run. It launches the
-  remote tasks and retrieves the results. The application can be run with
-  `python driver.py`.
-- [functions.py](functions.py) - This is the file that defines the remote
-  functions (in this case, just `train_cnn`).
-- [worker.py](worker.py) - This is the Python code that each worker process
-  runs. It imports the relevant modules and tells the scheduler what functions
-  it knows how to execute. Then it enters a loop that waits to receive tasks
-  from the scheduler.
-
-First, let's turn `train_cnn` into a remote function in Ray by writing it as
-follows. In this example application, a slightly more complicated version of
-this remote function is defined in [functions.py](functions.py).
+First, let's turn `train_cnn_and_compute_accuracy` into a remote function in Ray
+by writing it as follows. In this example application, a slightly more
+complicated version of this remote function is defined in
+[hyperopt.py](hyperopt.py).
 
 ```python
-@ray.remote([dict], [float])
-def train_cnn(hyperparameters):
-  # hyperparameters is a dictionary with keys
-  #   - "learning_rate"
-  #   - "batch_size"
-  #   - "dropout"
-  #   - "stddev"
-  # Train a deep network with the above hyperparameters
+@ray.remote([dict, np.ndarray, np.ndarray, np.ndarray, np.ndarray], [float])
+def train_cnn_and_compute_accuracy(hyperparameters, train_images, train_labels, validation_images, validation_labels):
+  # Actual work omitted.
   return validation_accuracy
 ```
 
 The only difference is that we added the `@ray.remote` decorator specifying a
-little bit of type information (the input is a dictionary and the return value
-is a float).
+little bit of type information (the input is a dictionary along with some numpy
+arrays, and the return value is a float).
 
-Now a call to `train_cnn` does not execute the function. It submits the task to
-the scheduler and returns an object reference for the output of the eventual
-computation. The scheduler, at its leisure, will schedule the task on a worker
-(which may live on the same machine or on a different machine in the cluster).
+Now a call to `train_cnn_and_compute_accuracy` does not execute the function. It
+submits the task to the scheduler and returns an object reference for the output
+of the eventual computation. The scheduler, at its leisure, will schedule the
+task on a worker (which may live on the same machine or on a different machine
+in the cluster).
 
 Now the for loop runs almost instantaneously because it does not do any actual
 computation. Instead, it simply submits a number of tasks to the scheduler.
@@ -119,28 +104,22 @@ computation. Instead, it simply submits a number of tasks to the scheduler.
 ```python
 result_refs = []
 for _ in range(100):
-  randparams = generate_random_params()
-  results.append((randparams, train_cnn(randparams, epochs)))
+  params = generate_random_params()
+  results.append((params, train_cnn_and_compute_accuracy(params, epochs)))
 ```
 
 If we wish to wait until the results have all been retrieved, we can retrieve
 their values with `ray.get`.
 
 ```python
-results = [(randparams, ray.get(ref)) for (randparams, ref) in result_refs]
-```
-
-This application can be run as follows.
-
-```
-python driver.py
+results = [(params, ray.get(ref)) for (params, ref) in result_refs]
 ```
 
 ### Additional notes
 
 **Early Stopping:** Sometimes when running an optimization, it is clear early on
 that the hyperparameters being used are bad (for example, the loss function may
-start diverging). In these situations, it makes sense to end that particular
-run early to save resources. This is implemented within the remote function
-`train_cnn`. If it detects that the optimization is going poorly, it returns
-early.
+start diverging). In these situations, it makes sense to end that particular run
+early to save resources. This is implemented within the remote function
+`train_cnn_and_compute_accuracy`. If it detects that the optimization is going
+poorly, it returns early.
diff --git a/examples/hyperopt/driver.py b/examples/hyperopt/driver.py
index 18ea78a15..d40a46f11 100644
--- a/examples/hyperopt/driver.py
+++ b/examples/hyperopt/driver.py
@@ -1,37 +1,62 @@
+# Most of the tensorflow code is adapted from Tensorflow's tutorial on using CNNs to train MNIST
+# https://www.tensorflow.org/versions/r0.9/tutorials/mnist/pros/index.html#build-a-multilayer-convolutional-network
 import numpy as np
 import ray
 import os
 
-import functions
+import tensorflow as tf
+from tensorflow.examples.tutorials.mnist import input_data
 
-num_workers = 3
-samples = 50
-epochs = 100
+import hyperopt
 
-worker_dir = os.path.dirname(os.path.abspath(__file__))
-worker_path = os.path.join(worker_dir, "worker.py")
-ray.services.start_ray_local(num_workers=num_workers, worker_path=worker_path)
+if __name__ == "__main__":
+  ray.services.start_ray_local(num_workers=3)
 
-best_params = None
-best_accuracy = 0
+  # The number of sets of random hyperparameters to try.
+  trials = 2
+  # The number of training passes over the dataset to use for network.
+  epochs = 10
 
-results = []
+  # Load the mnist data and turn the data into remote objects.
+  print "Downloading the MNIST dataset. This may take a minute."
+  mnist = input_data.read_data_sets("MNIST_data", one_hot=True)
+  train_images = ray.put(mnist.train.images)
+  train_labels = ray.put(mnist.train.labels)
+  validation_images = ray.put(mnist.validation.images)
+  validation_labels = ray.put(mnist.validation.labels)
 
-for i in range(samples):
-  learning_rate = 10 ** np.random.uniform(-6, 1)
-  batch_size = np.random.randint(30, 100)
-  dropout = np.random.uniform(0, 1)
-  stddev = 10 ** np.random.uniform(-3, 1)
-  randparams = {"learning_rate": learning_rate, "batch_size": batch_size, "dropout": dropout, "stddev": stddev}
-  results.append((randparams, functions.train_cnn(randparams, epochs)))
+  # Store the best parameters, the best accuracy, and all of the results.
+  best_params = None
+  best_accuracy = 0
+  results = []
 
-for i in range(samples):
-  params, ref = results[i]
-  accuracy = ray.get(ref)
-  print "With hyperparameters {}, we achieve an accuracy of {:.4}%.".format(params, 100 * accuracy)
-  if accuracy > best_accuracy:
-    best_params = params
-    best_accuracy = accuracy
-    print "Best parameters are now {}.".format(params)
+  # Randomly generate some hyperparameters, and launch a task for each set.
+  for i in range(trials):
+    learning_rate = 10 ** np.random.uniform(-5, 5)
+    batch_size = np.random.randint(1, 100)
+    dropout = np.random.uniform(0, 1)
+    stddev = 10 ** np.random.uniform(-5, 5)
+    params = {"learning_rate": learning_rate, "batch_size": batch_size, "dropout": dropout, "stddev": stddev}
+    results.append((params, hyperopt.train_cnn_and_compute_accuracy(params, epochs, train_images, train_labels, validation_images, validation_labels)))
 
-print "Best parameters over {} samples was {}, with an accuracy of {:.4}%.".format(samples, best_params, 100 * best_accuracy)
+  # Fetch the results of the tasks and print the results.
+  for i in range(trials):
+    params, ref = results[i]
+    accuracy = ray.get(ref)
+    print """We achieve accuracy {:.3}% with
+        learning_rate: {:.2}
+        batch_size: {}
+        dropout: {:.2}
+        stddev: {:.2}
+      """.format(100 * accuracy, params["learning_rate"], params["batch_size"], params["dropout"], params["stddev"])
+    if accuracy > best_accuracy:
+      best_params = params
+      best_accuracy = accuracy
+
+  # Record the best performing set of hyperparameters.
+  print """Best accuracy over {} trials was {:.3} with
+        learning_rate: {:.2}
+        batch_size: {}
+        dropout: {:.2}
+        stddev: {:.2}
+    """.format(trials, 100 * best_accuracy, best_params["learning_rate"], best_params["batch_size"], best_params["dropout"], best_params["stddev"])
diff --git a/examples/hyperopt/functions.py b/examples/hyperopt/hyperopt.py
similarity index 53%
rename from examples/hyperopt/functions.py
rename to examples/hyperopt/hyperopt.py
index bd58229ea..5ff1b8ae2 100644
--- a/examples/hyperopt/functions.py
+++ b/examples/hyperopt/hyperopt.py
@@ -1,11 +1,14 @@
-# Most of the tensorflow code is adapted from Tensorflow's tutorial on using CNNs to train MNIST
-# https://www.tensorflow.org/versions/r0.9/tutorials/mnist/pros/index.html#build-a-multilayer-convolutional-network
-import tensorflow as tf
-from tensorflow.examples.tutorials.mnist import input_data
-import numpy as np
 import ray
+import numpy as np
+import tensorflow as tf
 
-mnist = input_data.read_data_sets("MNIST_data", one_hot=True)
+def get_batch(data, batch_index, batch_size):
+  # This method currently drops data when num_data is not divisible by
+  # batch_size.
+  num_data = data.shape[0]
+  num_batches = num_data / batch_size
+  batch_index %= num_batches
+  return data[(batch_index * batch_size):((batch_index + 1) * batch_size)]
 
 def weight(shape, stddev):
   initial = tf.truncated_normal(shape, stddev=stddev)
@@ -21,29 +24,6 @@ def conv2d(x, W):
 def max_pool_2x2(x):
   return tf.nn.max_pool(x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding="SAME")
 
-@ray.remote([dict, int], [float])
-def train_cnn(params, epochs):
-  learning_rate = params["learning_rate"]
-  batch_size = params["batch_size"]
-  keep = 1 - params["dropout"]
-  stddev = params["stddev"]
-  x = tf.placeholder(tf.float32, shape=[None, 784])
-  y = tf.placeholder(tf.float32, shape=[None, 10])
-  keep_prob = tf.placeholder(tf.float32)
-  train_step, accuracy = cnn_setup(x, y, keep_prob, learning_rate, stddev)
-  with tf.Session() as sess:
-    sess.run(tf.initialize_all_variables())
-    for i in range(1, epochs):
-      batch = mnist.train.next_batch(batch_size)
-      sess.run(train_step, feed_dict={x: batch[0], y: batch[1], keep_prob: keep})
-      if i % 100 == 0: # checks if accuracy is low enough to stop early every set number of epochs
-        train_ac = accuracy.eval(feed_dict={x: batch[0], y: batch[1], keep_prob: 1.0})
-        if train_ac < 0.25: # Accuracy threshold is on a application to application basis.
-          totalacc = accuracy.eval(feed_dict={x: mnist.validation.images, y: mnist.validation.labels, keep_prob: 1.0})
-          return totalacc
-    totalacc = accuracy.eval(feed_dict={x: mnist.validation.images, y: mnist.validation.labels, keep_prob: 1.0})
-  return totalacc.astype("float64")
-
 def cnn_setup(x, y, keep_prob, lr, stddev):
   first_hidden = 32
   second_hidden = 64
@@ -68,3 +48,41 @@ def cnn_setup(x, y, keep_prob, lr, stddev):
   cross_entropy = tf.reduce_mean(-tf.reduce_sum(y * tf.log(y_conv), reduction_indices=[1]))
   correct_pred = tf.equal(tf.argmax(y_conv, 1), tf.argmax(y, 1))
   return tf.train.AdamOptimizer(lr).minimize(cross_entropy), tf.reduce_mean(tf.cast(correct_pred, tf.float32))
+
+# Define a remote function that takes a set of hyperparameters as well as the
+# data, consructs and trains a network, and returns the validation accuracy.
+@ray.remote([dict, int, np.ndarray, np.ndarray, np.ndarray, np.ndarray], [float])
+def train_cnn_and_compute_accuracy(params, epochs, train_images, train_labels, validation_images, validation_labels):
+  # Extract the hyperparameters from the params dictionary.
+  learning_rate = params["learning_rate"]
+  batch_size = params["batch_size"]
+  keep = 1 - params["dropout"]
+  stddev = params["stddev"]
+  # Create the input placeholders for the network.
+  x = tf.placeholder(tf.float32, shape=[None, 784])
+  y = tf.placeholder(tf.float32, shape=[None, 10])
+  keep_prob = tf.placeholder(tf.float32)
+  # Create the network.
+  train_step, accuracy = cnn_setup(x, y, keep_prob, learning_rate, stddev)
+  # Do the training and evaluation.
+  with tf.Session() as sess:
+    # Initialize the network weights.
+    sess.run(tf.initialize_all_variables())
+    for i in range(1, epochs):
+      # Fetch the next batch of data.
+      image_batch = get_batch(train_images, i, batch_size)
+      label_batch = get_batch(train_labels, i, batch_size)
+      # Do one step of training.
+      sess.run(train_step, feed_dict={x: image_batch, y: label_batch, keep_prob: keep})
+      if i % 100 == 0:
+        # Estimate the training accuracy every once in a while.
+        train_ac = accuracy.eval(feed_dict={x: image_batch, y: label_batch, keep_prob: 1.0})
+        # If the training accuracy is too low, stop early in order to avoid
+        # wasting computation.
+        if train_ac < 0.25:
+          # Compute the validation accuracy and return.
+          totalacc = accuracy.eval(feed_dict={x: validation_images, y: validation_labels, keep_prob: 1.0})
+          return totalacc
+    # Training is done, compute the validation accuracy and return.
+    totalacc = accuracy.eval(feed_dict={x: validation_images, y: validation_labels, keep_prob: 1.0})
+  return float(totalacc)
diff --git a/examples/hyperopt/worker.py b/examples/hyperopt/worker.py
deleted file mode 100644
index 45310e526..000000000
--- a/examples/hyperopt/worker.py
+++ /dev/null
@@ -1,15 +0,0 @@
-import argparse
-import ray
-
-import functions
-
-parser = argparse.ArgumentParser(description="Parse addresses for the worker to connect to.")
-parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address")
-parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address")
-parser.add_argument("--worker-address", default="127.0.0.1:40001", type=str, help="the worker's address")
-
-if __name__ == "__main__":
-  args = parser.parse_args()
-  ray.connect(args.scheduler_address, args.objstore_address, args.worker_address)
-  ray.register_module(functions)
-  ray.worker.main_loop()