Training on Cloud AI Platform

Learning Objectives

Use CAIP to run a distributed training job

Introduction

After having testing our training pipeline both locally and in the cloud on a susbset of the data, we can submit another (much larger) training job to the cloud. It is also a good idea to run a hyperparameter tuning job to make sure we have optimized the hyperparameters of our model.

This notebook illustrates how to do distributed training and hyperparameter tuning on Cloud AI Platform.

To start, we'll set up our environment variables as before.



In [ ]:

    
PROJECT = "cloud-training-demos"  # Replace with your PROJECT
BUCKET = "cloud-training-bucket"  # Replace with your BUCKET

REGION = "us-central1"            # Choose an available region for Cloud CAIP
TFVERSION = "1.14"                # TF version for CMLE to use



In [ ]:

    
import os
os.environ["BUCKET"] = BUCKET
os.environ["PROJECT"] = PROJECT
os.environ["REGION"] = REGION
os.environ["TFVERSION"] = TFVERSION



In [ ]:

    
%%bash
gcloud config set project $PROJECT
gcloud config set compute/region $REGION

Next, we'll look for the preprocessed data for the babyweight model and copy it over if it's not there.



In [ ]:

    
%%bash
if ! gsutil ls -r gs://$BUCKET | grep -q gs://$BUCKET/babyweight/preproc; then
    gsutil mb -l ${REGION} gs://${BUCKET}
    # copy canonical set of preprocessed files if you didn't do previous notebook
    gsutil -m cp -R gs://cloud-training-demos/babyweight gs://${BUCKET}
fi



In [ ]:

    
%%bash
gsutil ls gs://${BUCKET}/babyweight/preproc/*-00000*

In the previous labs we developed our TensorFlow model and got it working on a subset of the data. Now we can package the TensorFlow code up as a Python module and train it on Cloud AI Platform.

Train on Cloud AI Platform

Training on Cloud AI Platform requires two things:

Configuring our code as a Python package
Using gcloud to submit the training code to Cloud AI Platform

Move code into a Python package

A Python package is simply a collection of one or more .py files along with an __init__.py file to identify the containing directory as a package. The __init__.py sometimes contains initialization code but for our purposes an empty file suffices.

The bash command touch creates an empty file in the specified location, the directory babyweight should already exist.



In [ ]:

    
%%bash
touch babyweight/trainer/__init__.py

We then use the %%writefile magic to write the contents of the cell below to a file called task.py in the babyweight/trainer folder.



In [ ]:

    
%%writefile babyweight/trainer/task.py
import argparse
import json
import os

from . import model

import tensorflow as tf


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--bucket",
        help="GCS path to data. We assume that data is in \
        gs://BUCKET/babyweight/preproc/",
        required=True
    )
    parser.add_argument(
        "--output_dir",
        help="GCS location to write checkpoints and export models",
        required=True
    )
    parser.add_argument(
        "--batch_size",
        help="Number of examples to compute gradient over.",
        type=int,
        default=512
    )
    parser.add_argument(
        "--job-dir",
        help="this model ignores this field, but it is required by gcloud",
        default="junk"
    )
    parser.add_argument(
        "--nnsize",
        help="Hidden layer sizes to use for DNN feature columns -- provide \
        space-separated layers",
        nargs="+",
        type=int,
        default=[128, 32, 4]
    )
    parser.add_argument(
        "--nembeds",
        help="Embedding size of a cross of n key real-valued parameters",
        type=int,
        default=3
    )
    parser.add_argument(
        "--train_examples",
        help="Number of examples (in thousands) to run the training job over. \
        If this is more than actual \
        So specifying 1000 here when you have only 100k examples \
        makes this 10 epochs.",
        type=int,
        default=5000
    )
    parser.add_argument(
        "--pattern",
        help="Specify a pattern that has to be in input files. \
        For example 00001-of \
        will process only one shard",
        default="of"
    )
    parser.add_argument(
        "--eval_steps",
        help="Positive number of steps for which to evaluate model. \
        Default to None, which means to evaluate until \
        input_fn raises an end-of-input exception",
        type=int,
        default=None
    )

    # Parse arguments
    args = parser.parse_args()
    arguments = args.__dict__

    # Pop unnecessary args needed for gcloud
    arguments.pop("job-dir", None)

    # Assign the arguments to the model variables
    output_dir = arguments.pop("output_dir")
    model.BUCKET = arguments.pop("bucket")
    model.BATCH_SIZE = arguments.pop("batch_size")
    model.TRAIN_STEPS = (
        arguments.pop("train_examples") * 1000) / model.BATCH_SIZE
    model.EVAL_STEPS = arguments.pop("eval_steps")
    print ("Will train for {} steps using batch_size={}".format(
        model.TRAIN_STEPS, model.BATCH_SIZE))
    model.PATTERN = arguments.pop("pattern")
    model.NEMBEDS = arguments.pop("nembeds")
    model.NNSIZE = arguments.pop("nnsize")
    print ("Will use DNN size of {}".format(model.NNSIZE))

    # Append trial_id to path if we are doing hptuning
    # This code can be removed if you are not using hyperparameter tuning
    output_dir = os.path.join(
        output_dir,
        json.loads(
            os.environ.get("TF_CONFIG", "{}")
        ).get("task", {}).get("trial", "")
    )

    # Run the training job
    model.train_and_evaluate(output_dir)

In the same way we can write to the file model.py the model that we developed in the previous notebooks.



In [ ]:

    
%%writefile babyweight/trainer/model.py
import shutil

import numpy as np
import tensorflow as tf

tf.logging.set_verbosity(tf.logging.INFO)

BUCKET = None  # set from task.py
PATTERN = "of"

CSV_COLUMNS = [
    "weight_pounds",
    "is_male",
    "mother_age",
    "plurality",
    "gestation_weeks",
]
LABEL_COLUMN = "weight_pounds"
DEFAULTS = [[0.0], ["null"], [0.0], ["null"], [0.0]]

TRAIN_STEPS = 10000
EVAL_STEPS = None
BATCH_SIZE = 512
NEMBEDS = 3
NNSIZE = [64, 16, 4]


def read_dataset(filename_pattern, mode, batch_size):

    def _input_fn():

        def decode_csv(value_column):
            columns = tf.decode_csv(
                records=value_column,
                record_defaults=DEFAULTS
            )
            features = dict(zip(CSV_COLUMNS, columns))
            label = features.pop(LABEL_COLUMN)
            return features, label

        file_path = "gs://{}/babyweight/preproc/{}*{}*".format(
            BUCKET, filename_pattern, PATTERN)
        file_list = tf.gfile.Glob(filename=file_path)

        dataset = (
            tf.data.TextLineDataset(filenames=file_list).map(
                map_func=decode_csv)
        )

        if mode == tf.estimator.ModeKeys.TRAIN:
            num_epochs = None  # indefinitely
            dataset = dataset.shuffle(buffer_size=10*batch_size)
        else:
            num_epochs = 1
        dataset = dataset.repeat(count=num_epochs).batch(batch_size=batch_size)

        return dataset

    return _input_fn


def get_wide_deep():

    fc_is_male = tf.feature_column.categorical_column_with_vocabulary_list(
        key="is_male",
        vocabulary_list=["True", "False", "Unknown"]
    )

    fc_plurality = tf.feature_column.categorical_column_with_vocabulary_list(
        key="plurality",
        vocabulary_list=[
            "Single(1)",
            "Twins(2)",
            "Triplets(3)",
            "Quadruplets(4)",
            "Quintuplets(5)",
            "Multiple(2+)"
        ]
    )

    fc_mother_age = tf.feature_column.numeric_column("mother_age")

    fc_gestation_weeks = tf.feature_column.numeric_column("gestation_weeks")

    fc_age_buckets = tf.feature_column.bucketized_column(
        source_column=fc_mother_age, 
        boundaries=np.arange(start=15, stop=45, step=1).tolist()
    )

    fc_gestation_buckets = tf.feature_column.bucketized_column(
        source_column=fc_gestation_weeks,
        boundaries=np.arange(start=17, stop=47, step=1).tolist())

    wide = [
        fc_is_male,
        fc_plurality,
        fc_age_buckets,
        fc_gestation_buckets
    ]

    # Feature cross all the wide columns and embed into a lower dimension
    crossed = tf.feature_column.crossed_column(
        keys=wide, hash_bucket_size=20000
    )
    fc_embed = tf.feature_column.embedding_column(
        categorical_column=crossed,
        dimension=3
    )

    # Continuous columns are deep, have a complex relationship with the output
    deep = [
        fc_mother_age,
        fc_gestation_weeks,
        fc_embed
    ]

    return wide, deep


def serving_input_fn():
    feature_placeholders = {
        "is_male": tf.placeholder(dtype=tf.string, shape=[None]),
        "mother_age": tf.placeholder(dtype=tf.float32, shape=[None]),
        "plurality": tf.placeholder(dtype=tf.string, shape=[None]),
        "gestation_weeks": tf.placeholder(dtype=tf.float32, shape=[None])
    }

    features = {
        key: tf.expand_dims(input=tensor, axis=-1)
        for key, tensor in feature_placeholders.items()
    }

    return tf.estimator.export.ServingInputReceiver(
        features=features, 
        receiver_tensors=feature_placeholders
    )


def my_rmse(labels, predictions):
    pred_values = predictions["predictions"]
    return {
        "rmse": tf.metrics.root_mean_squared_error(
            labels=labels,
            predictions=pred_values
        )
    }


def train_and_evaluate(output_dir):
    wide, deep = get_wide_deep()
    EVAL_INTERVAL = 300  # seconds

    run_config = tf.estimator.RunConfig(
        save_checkpoints_secs=EVAL_INTERVAL,
        keep_checkpoint_max=3)

    estimator = tf.estimator.DNNLinearCombinedRegressor(
        model_dir=output_dir,
        linear_feature_columns=wide,
        dnn_feature_columns=deep,
        dnn_hidden_units=NNSIZE,
        config=run_config)

    estimator = tf.contrib.estimator.add_metrics(estimator, my_rmse)

    train_spec = tf.estimator.TrainSpec(
        input_fn=read_dataset(
            "train", tf.estimator.ModeKeys.TRAIN, BATCH_SIZE),
        max_steps=TRAIN_STEPS)

    exporter = tf.estimator.LatestExporter(
        name="exporter",
        serving_input_receiver_fn=serving_input_fn,
        exports_to_keep=None)

    eval_spec = tf.estimator.EvalSpec(
        input_fn=read_dataset(
            "eval", tf.estimator.ModeKeys.EVAL, 2**15),
        steps=EVAL_STEPS,
        start_delay_secs=60,  # start evaluating after N seconds
        throttle_secs=EVAL_INTERVAL,  # evaluate every N seconds
        exporters=exporter)

    tf.estimator.train_and_evaluate(
        estimator=estimator,
        train_spec=train_spec,
        eval_spec=eval_spec
    )

Train locally

After moving the code to a package, make sure it works as a standalone. Note, we incorporated the --pattern and --train_examples flags so that we don't try to train on the entire dataset while we are developing our pipeline. Once we are sure that everything is working on a subset, we can change the pattern so that we can train on all the data. Even for this subset, this takes about 3 minutes in which you won't see any output ...



In [ ]:

    
%%bash
echo "bucket=$BUCKET"
rm -rf babyweight_trained
export PYTHONPATH=${PYTHONPATH}:${PWD}/babyweight
python -m trainer.task \
    --bucket=$BUCKET \
    --output_dir=babyweight_trained \
    --job-dir=./tmp \
    --pattern="00000-of-"\
    --train_examples=1 \
    --eval_steps=1

Making predictions

The JSON below represents an input into your prediction model. Write the input.json file below with the next cell, then run the prediction locally to assess whether it produces predictions correctly.



In [ ]:

    
%%writefile inputs.json
{"is_male": "True", "mother_age": 26.0, "plurality": "Single(1)", "gestation_weeks": 39}
{"is_male": "False", "mother_age": 26.0, "plurality": "Single(1)", "gestation_weeks": 39}



In [ ]:

    
%%bash
MODEL_LOCATION=$(ls -d $(pwd)/babyweight_trained/export/exporter/* | tail -1)
echo $MODEL_LOCATION
gcloud ml-engine local predict --model-dir=$MODEL_LOCATION --json-instances=inputs.json

Training on the Cloud with CAIP

Once the code works in standalone mode, you can run it on Cloud AI Platform. Because this is on the entire dataset, it will take a while. The training run took about an hour for me. You can monitor the job from the GCP console in the Cloud AI Platform section.



In [ ]:

    
%%bash
OUTDIR=gs://${BUCKET}/babyweight/trained_model
JOBNAME=babyweight_$(date -u +%y%m%d_%H%M%S)
echo $OUTDIR $REGION $JOBNAME
gsutil -m rm -rf $OUTDIR
gcloud ai-platform jobs submit training $JOBNAME \
    --region=$REGION \
    --module-name=trainer.task \
    --package-path=$(pwd)/babyweight/trainer \
    --job-dir=$OUTDIR \
    --staging-bucket=gs://$BUCKET \
    --scale-tier=STANDARD_1 \
    --runtime-version=$TFVERSION \
    -- \
    --bucket=${BUCKET} \
    --output_dir=${OUTDIR} \
    --train_examples=200000CAIP

When I ran it, I used train_examples=2000000. When training finished, I filtered in the Stackdriver log on the word "dict" and saw that the last line was:

Saving dict for global step 5714290: average_loss = 1.06473, global_step = 5714290, loss = 34882.4, rmse = 1.03186

The final RMSE was 1.03 pounds.

Hyperparameter tuning

All of these are command-line parameters to my program. To do hyperparameter tuning, create hyperparam.xml and pass it as --configFile. This step will take 1 hour -- you can increase maxParallelTrials or reduce maxTrials to get it done faster. Since maxParallelTrials is the number of initial seeds to start searching from, you don't want it to be too large; otherwise, all you have is a random search.



In [ ]:

    
%%writefile hyperparam.yaml
trainingInput:
    scaleTier: STANDARD_1
    hyperparameters:
        hyperparameterMetricTag: rmse
        goal: MINIMIZE
        maxTrials: 20
        maxParallelTrials: 5
        enableTrialEarlyStopping: True
        params:
        - parameterName: batch_size
          type: INTEGER
          minValue: 8
          maxValue: 512
          scaleType: UNIT_LOG_SCALE
        - parameterName: nembeds
          type: INTEGER
          minValue: 3
          maxValue: 30
          scaleType: UNIT_LINEAR_SCALE
        - parameterName: nnsize
          type: INTEGER
          minValue: 64
          maxValue: 512
          scaleType: UNIT_LOG_SCALE



In [ ]:

    
%%bash
OUTDIR=gs://${BUCKET}/babyweight/hyperparam
JOBNAME=babyweight_$(date -u +%y%m%d_%H%M%S)
echo $OUTDIR $REGION $JOBNAME
gsutil -m rm -rf $OUTDIR
gcloud ai-platform jobs submit training $JOBNAME \
    --region=$REGION \
    --module-name=trainer.task \
    --package-path=$(pwd)/babyweight/trainer \
    --job-dir=$OUTDIR \
    --staging-bucket=gs://$BUCKET \
    --scale-tier=STANDARD_1 \
    --config=hyperparam.yaml \
    --runtime-version=$TFVERSION \
    -- \
    --bucket=${BUCKET} \
    --output_dir=${OUTDIR} \
    --eval_steps=10 \
    --train_examples=20000

Repeat training

Now that we've determined the optimal hyparameters, we'll retrain with these tuned parameters. Note the last line.



In [ ]:

    
%%bash
OUTDIR=gs://${BUCKET}/babyweight/trained_model_tuned
JOBNAME=babyweight_$(date -u +%y%m%d_%H%M%S)
echo $OUTDIR $REGION $JOBNAME
gsutil -m rm -rf $OUTDIR
gcloud ai-platform jobs submit training $JOBNAME \
    --region=$REGION \
    --module-name=trainer.task \
    --package-path=$(pwd)/babyweight/trainer \
    --job-dir=$OUTDIR \
    --staging-bucket=gs://$BUCKET \
    --scale-tier=STANDARD_1 \
    --runtime-version=$TFVERSION \
    -- \
    --bucket=${BUCKET} \
    --output_dir=${OUTDIR} \
    --train_examples=20000 --batch_size=35 --nembeds=16 --nnsize=281

Copyright 2017 Google Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License