MNIST and TensorFlow

Running the Disdat MNIST Pipeline

We've shamelessly adapted the MNIST Tensorflow example. Here we've broken the example down into three steps (Disdat Tasks) all of which are in $DISDAT_HOME/examples/ex_pipelines/mnist.py.

  1. Get MNIST data: class GetDataGz(PipeTask) This downloads four gzip files and stores them in a bundle called MNIST.data.gz

  2. Train the model: class Train(PipeTask)This PipeTask depends on the GetDataGz tasks, gets the gzip files, builds a Tensorflow graph and trains it. It stores the saved model into an output bundle called MNIST.trained.

  3. Evaluate accuracy: class Evaluate(PipeTask): This PipeTask depends on both upstream tasks. It rebuilds the TF graph, restores the values, and evaluates the model. It returns a single accuracy float in its output bundle MNIST.eval

Train the classifier

$dsdt apply pipelines.mnist.Train
INFO: Informed scheduler that task   DriverTask_False______48a9755ee1   has status   PENDING
INFO: Informed scheduler that task   Train__99914b932b   has status   PENDING
INFO: Informed scheduler that task   GetDataGz__99914b932b   has status   PENDING
INFO: Done scheduling tasks
[... more output ...]
Successfully downloaded train-images-idx3-ubyte.gz 9912422 bytes.
Successfully downloaded train-labels-idx1-ubyte.gz 28881 bytes.
Successfully downloaded t10k-images-idx3-ubyte.gz 1648877 bytes.
Successfully downloaded t10k-labels-idx1-ubyte.gz 4542 bytes.
[... more output ...]

End training.
===== Luigi Execution Summary =====

Scheduled 3 tasks of which:
* 3 ran successfully:
    - 1 DriverTask(...)
    - 1 GetDataGz(...)
    - 1 Train(...)

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

NOTE: we could have run just the first step by:

dsdt apply pipelines.mnist.GetDataGz

and all three steps by:

dsdt apply pipelines.mnist.Evaluate

Inspect the training data and trained model

This produces two bundles: MNIST.data.gz and MNIST.trained.

The training data bundle presents as a dictionary, with the key signifying what kind of information the value (a file link) holds. 

$dsdt cat MNIST.data.gz
{'train-images': ['/Users/kyocum/.disdat/context/examples/objects/10beb5e2-3c07-4a4c-a1f7-3c2b0e2407f7/train-images-idx3-ubyte.gz'], 
'train-labels': ['/Users/kyocum/.disdat/context/examples/objects/10beb5e2-3c07-4a4c-a1f7-3c2b0e2407f7/train-labels-idx1-ubyte.gz'], 
't10k-images': ['/Users/kyocum/.disdat/context/examples/objects/10beb5e2-3c07-4a4c-a1f7-3c2b0e2407f7/t10k-images-idx3-ubyte.gz'], 
't10k-labels': ['/Users/kyocum/.disdat/context/examples/objects/10beb5e2-3c07-4a4c-a1f7-3c2b0e2407f7/t10k-labels-idx1-ubyte.gz']}

The trained model also presents as a dictionary, with a key save_files pointing to a list of saved output files (note you can have directories in your bundle). And another key model_name which just holds the name of the saved model. 

$dsdt cat MNIST.trained
{'save_files': array(['/Users/kyocum/.disdat/context/examples/objects/53f75bd7-6841-47ca-8f8b-904c5f02dada/MNIST/MNIST_tf_model.index',
       '/Users/kyocum/.disdat/context/examples/objects/53f75bd7-6841-47ca-8f8b-904c5f02dada/MNIST/checkpoint',
       '/Users/kyocum/.disdat/context/examples/objects/53f75bd7-6841-47ca-8f8b-904c5f02dada/MNIST/MNIST_tf_model.data-00000-of-00001',
       '/Users/kyocum/.disdat/context/examples/objects/53f75bd7-6841-47ca-8f8b-904c5f02dada/MNIST/MNIST_tf_model.meta'],
      dtype='<U124'), 'model_name': ['MNIST_tf_model']}

Evaluate the trained model

Let's run the lst stage of the pipeline. Note that under the hood, Luigi is telling us that the GetDataGz and Train tasks are done. 

$dsdt apply -f pipelines.mnist.Evaluate
INFO: Informed scheduler that task   DriverTask_False______a33b532760   has status   PENDING
INFO: Informed scheduler that task   Evaluate__99914b932b   has status   PENDING
INFO: Informed scheduler that task   GetDataGz__99914b932b   has status   DONE
INFO: Informed scheduler that task   Train__99914b932b   has status   DONE
[ ... ]
===== Luigi Execution Summary =====

Scheduled 4 tasks of which:
* 2 complete ones were encountered:
    - 1 GetDataGz(...)
    - 1 Train(...)
* 2 ran successfully:
    - 1 DriverTask(...)
    - 1 Evaluate(...)

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

Now you've produced three bundles. Let's look at evaluate task's final output bundle:

$dsdt cat MNIST.eval
0.9203000068664551

Pipeline Detail

Setting up dependencies

Let's look at the Train task. Recall that a Disdat PipeTask consists of two functions: pipe_requires and pipe_run. Starting with pipe_requires()

class Train(PipeTask):
    """ Pipe Task 2
    Train the softmax layer.
    Returns:
        (dict): all the model files in 'save_files' and the name of the dir in 'save_dir'
    """

    def pipe_requires(self):
        """ Depend on the gzip files being downloaded  """
        self.add_dependency("input_gzs", GetDataGz, {})
        self.set_bundle_name("MNIST.trained")
...

  • pipe_requires: This declares the tasks that must run before this task. The statement self.add_dependency("input_gzs", GetDataGz, params={}) says that the current task needs a GetDataGz instance to run with no parameters. It also says that Disdat should setup the output of that task as a named parameter to pipe_run called 'input_gzs'.

Outputs

Next lets look into pipe_run in more detail. Note that TensorFlow code is just Python TensorFlow code. No magic. Let's focus on lines 19 and 30. This will illustrate how tasks save state (via files) and how you implicitly tell Disdat to put those files in a bundle. 

    def pipe_run(self, input_gzs=None):
        print("Beginning training . . . ")
    
        # Bring the data in as an mnist tutorial dataset
        mnist = convert_to_data_sets(input_gzs, one_hot=True)
    
        g = tf.Graph()
        with g.as_default():
            train_step, cross_entropy, x, y_, _ = make_model()
            saver = tf.train.Saver()
    
        with tf.Session(graph=g) as sess:
            sess.run(tf.global_variables_initializer())
    
            for _ in range(1000):
                batch = mnist.train.next_batch(100)
                train_step.run(feed_dict={x: batch[0], y_: batch[1]})
    
----------->save_dir = os.path.join(self.get_output_dir(), 'MNIST')
            saver.save(sess, os.path.join(save_dir, 'MNIST_tf_model'))
    
        print("End training.")
    
        # Note 1: When returning a dictionary, disdat requires you to use a sequence
        # in the value.
    
        # Note 2: You can return file paths, directories, or luigi.LocalTarget.
        # If a directory, Disdat takes all files directly under the directory.
    
------->return {'save_files': [save_dir], 'model_name': ['MNIST_tf_model']}

Bundles are designed primarily to wrap other data formats -- not re-invent them. Thus tasks typically produce one or more files as output (in whatever format they choose), and they pass the names of those files directly as return values or place them in lists, dictionaries, tuples, or simple Pandas dataframes.

However, Disdat manages your output paths for you -- you just need to name the files, not describe outputs with a cascade of custom directory names. That is, instead of: /Users/moonga/created/12_31_18/input_data/11_18_18/models/results.mdl you only provide results.mdl You can do this by:

  • Calling self.create_output_file("results.mdl"): this returns a luigi.LocalTarget object. You can open and write to it.

  • Or, like line 19 above, you can call self.get_output_dir(): This returns a fully-qualified path to your bundle's output directory. You can place files directly into this directory. This is handy because some libraries, like TensorFlow's Saver object, just want an output directory -- it's fragile and difficult to enumerate all the outputs ahead of time.

Note: You can always just create files wherever you wish, i.e., on your local machine or in S3. And you can place the fully qualified paths in your output bundle. In this case, Disdat will copy-in your external files into your data context.

The benefit of using create_output_file and get_output_dir is that there is no data copying when Disdat creates the bundle.

When you're done making files, you still need to return the file paths (if you want them to be in your output bundle). You can return:

  • The luigi.LocalTarget: Disdat knows how to interpet them.

  • The full paths of any file, e.g., os.path.join(self.get_output_dir(), "my_results.txt")

  • A directory (line 30 above): Disdat will include files in a sub-directory automatically (one-level deep).

In Disdat, the pipe_run function may return scalars, lists, dictionaries, tuples, Numpy ndarrays, or Pandas DataFrames (see Task Return Types). Downstream tasks will be called with the same type. Under the hood, Disdat bundles those data types, storing them as Protocol Buffers in a managed directory (typically ~/.disdat).

Pushing outputs to S3 (optional)

Finally, let's say that you're ready to share the training data, model, and results. then we are going to commit each of our output bundles. Committing is simply setting a flag that tells Disdat that you like this version so much, you're willing to put it up on S3. 

$ dsdt commit MNIST.eval; dsdt commit MNIST.data.gz; dsdt commit MNIST.trained

Now let's push each bundle up to our remote context: 

$ dsdt push MNIST.eval; dsdt push MNIST.data.gz; dsdt push MNIST.trained

Now all of your data is safely on S3. To illustrate, let's delete our local copies and pull it back. Note that we are using the -l flag to tell Disdat to download (localize) our data from S3 to the local file system. 

$ dsdt rm --all MNI.*
$ dsdt pull -l MNIST.eval; dsdt pull -l MNIST.data.gz; dsdt pull -l MNIST.trained

You can see all of your bundles using dsdt ls -v

Dockerize MNIST

Change to the examples directory. This is what contains the examples pipeline's setup.py file. Let's create the container and push it up to AWS ECR as well:

$cd $CODE
$dsdt dockerize --push .

Assuming you have set up your AWS Batch queue in the Disdat configuration file, go ahead and run the container.

$dsdt run --backend AWSBatch . pipelines.mnist.Evaluate

Check your AWS Batch queue to see the job progress from ready to finished. Then pull the bundles and list out the committed (-c) bundles that begin with MNIST.*

$dsdt pull
$dsdt ls -vc MNIST.*
NAME                	PROC_NAME           	OWNER   	DATE              	COMMITTED	UUID                                    	TAGS
MNIST.eval          	Evaluate__99914b932b	root    	02-01-20 11:14:32 	True    	22eb7d23-ee8b-4985-87d6-fa05dc3aea53
MNIST.trained       	Train__99914b932b   	root    	02-01-20 11:14:31 	True    	3ea79fb6-9641-499f-9cc1-89ed23740e8e
MNIST.data.gz       	GetDataGz__99914b932b	root    	02-01-20 11:14:28 	True    	84d6f033-d94c-4a75-95b6-d7b4681bdad5

How to run Disdat AWS Batch tasks:

Default: By default Disdat will user your credentials in ~/.aws/credentials. If they are standard credentials (not session tokens), then Disdat will create a token (expiring in 43200 seconds) to give your container sufficient privileges.

AWS session token: In the case that Disdat finds a session token in your credentials file, it will submit the job with that token. If your org has a short time limit, then your jobs might fail when they access AWS resources. Typically this happens when Disdat writes your results back to S3. Bummer.

AWS IAM roles: To avoid short tokens, create a role using IAM, and pass it in to the dsdt run command. IAM is a pain, but if you don't have an SRE to help you here, you should read up on roles.

Creating a role:

  • Go to IAM on the AWS Console and choose create role.

  • Select AWS Service as the trusted entity

  • Choose Elastic Container Service as your service use case

  • Choose Elastic Container Service Task

  • Hit next for permissions

  • Search for S3 and then choose AmazonS3FullAccess

  • Hit tags, and enter some tags. I like ‘owner’ and ‘project’ at a minimum

  • Give your role a name and description

  • At the end you should see your role, copy the ARN to your clipboard

dsdt run --job-role-arn <YOUR ROLE ARN> --backend AWSBatch . pipelines.dependent_tasks.B

PreviousExamplesNextSpacy Task

Last updated

Was this helpful?