<jupyter_start><jupyter_text>OverviewIn this notebook, you'll learn how to submit a job to AI Platform Training. In the job you'll train your TensorFlow 2 model and export the saved model to Cloud Storage. Dataset[CTA - Ridership - Daily Boarding Totals](https://data.cityofchicago.org/Transportation/CTA-Ridership-Daily-Boarding-Totals/6iiy-9s97): This dataset shows systemwide boardings for both bus and rail services provided by Chicago Transit Authority, dating back to 2001. ObjectiveThe goal is to forecast future transit ridership in the City of Chicago, based on previous ridership.  Install packages and dependencies<jupyter_code>%%capture
!pip uninstall -y google-cloud-aiplatform
!pip install --user git+https://github.com/googleapis/python-aiplatform.git@dev-test
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)<jupyter_output><empty_output><jupyter_text>Import libraries and define constants<jupyter_code>import datetime
import googleapiclient.discovery
import os
import time

import numpy as np
import pandas as pd
import tensorflow as tf

from google.api_core.client_options import ClientOptions
from google.cloud import storage
from sklearn.preprocessing import StandardScaler
# Check the TensorFlow version installed

tf.__version__<jupyter_output><empty_output><jupyter_text>Setup AI Platform SDK workspace<jupyter_code>from google.cloud import aiplatform
import uuid
PROJECT_ID = !gcloud config list --format 'value(core.project)'
# Enter your project, region, and a bucket name. Then run the  cell to make sure the
# Cloud SDK uses the right project for all the commands in this notebook.

PROJECT = PROJECT_ID[0] # REPLACE WITH YOUR PROJECT ID 
BUCKET = 'cloud-ai-platform-2f444b6a-a742-444b-b91a-c7519f51bd77' # REPLACE WITH A UNIQUE BUCKET NAME e.g. your PROJECT NAME
REGION = 'us-central1' # REPLACE WITH YOUR BUCKET REGION e.g. us-central1
BUCKET_URI = 'gs://' + BUCKET

#Don't change the following command - this is to check if you have changed the project name above.
assert PROJECT != 'your-project-name', 'Don''t forget to change the project variables!'<jupyter_output><empty_output><jupyter_text>**Initialize the client for AI Platform (Unified).**<jupyter_code>#aiplatform.constants.API_BASE_PATH = "staging-aiplatform.sandbox.googleapis.com" # set api endpoint to staging
aiplatform.constants.API_BASE_PATH = "aiplatform.googleapis.com"<jupyter_output><empty_output><jupyter_text>**Initialize Model Builder SDK and set an Experiment**<jupyter_code>MY_PROJECT = PROJECT # Please change this to your own test project
MY_STAGING_BUCKET = 'gs://cloud-ai-platform-2f444b6a-a742-444b-b91a-c7519f51bd77'  # bucket should be in same region as ucaip
BUCKET = MY_STAGING_BUCKET
aiplatform.init(project=MY_PROJECT,
                experiment="forecasting-experiment",
                staging_bucket=MY_STAGING_BUCKET)<jupyter_output><empty_output><jupyter_text>Define Constants<jupyter_code># Dataset parameters

target_col = 'total_rides' # The variable you are predicting
ts_col = 'service_date' # The name of the column with the date field
# Model parameters

freq = 'D' # Daily frequency
n_input_steps = 30 # Lookback window
n_output_steps = 7 # How many steps to predict forward
n_seasons = 7 # Monthly periodicity

train_split = 0.8 # % Split between train/test data
epochs = 1000 # How many passes through the data (early-stopping will cause training to stop before this)
patience = 5 # Terminate training after the validation loss does not decrease after this many epochs

lstm_units = 64
input_layer_name = 'lstm_input'
# Training parameters

MODEL_NAME = 'cta_ridership'
FRAMEWORK='TENSORFLOW'
RUNTIME_VERSION = '2.4'
PYTHON_VERSION = '3.7'
PREDICTIONS_FILE = 'predictions.json'<jupyter_output><empty_output><jupyter_text>Create a Cloud Storage bucket**The following steps are required, regardless of your notebook environment.**When you submit a training job using the Cloud SDK, you upload a Python packagecontaining your training code to a Cloud Storage bucket. AI Platform runsthe code from this package. In this tutorial, AI Platform also saves thetrained model that results from your job in the same bucket. You can thencreate an AI Platform model version based on this output in order to serveonline predictions.<jupyter_code>storage_client = storage.Client()
try:
    bucket = storage_client.get_bucket(BUCKET)
    print('Bucket exists, let''s not recreate it.')
except:
    bucket = storage_client.create_bucket(BUCKET)
    print('Created bucket: ' + BUCKET)<jupyter_output>Bucket exists, lets not recreate it.<jupyter_text>Load and preview the dataPre-processing on the original dataset has been done for you and made available on Cloud Storage.<jupyter_code>processed_file = 'cta_ridership.csv' # Which file to save the results to

if os.path.exists(processed_file):
    input_file = processed_file # File created in previous lab
else:
    input_file = f'data/{processed_file}'
df = pd.read_csv(input_file, index_col=ts_col, parse_dates=True)

# Plot 30 days of ridership 
_ = df[target_col][:30].plot()
# Define some characteristics of the data that will be used later
n_features = len(df.columns)

# Index of target column. Used later when creating dataframes.
target_col_num = df.columns.get_loc(target_col)<jupyter_output><empty_output><jupyter_text>Process data<jupyter_code># Split data

size = int(len(df) * train_split)
df_train, df_test = df[0:size].copy(deep=True), df[size:len(df)].copy(deep=True)

df_train.head()
_ = df_train.plot()<jupyter_output><empty_output><jupyter_text>Scale values<jupyter_code># Review original values

df_train.head()
# For neural networks to converge quicker, it is helpful to scale the values.
# For example, each feature might be transformed to have a mean of 0 and std. dev. of 1.
#
# You are working with a mix of features, input timesteps, output horizon, etc.
# which don't work out-of-the-box with common scaling utilities.
# So, here are a couple wrappers to handle scaling and inverting the scaling.

feature_scaler = StandardScaler()
target_scaler = StandardScaler()

def scale(df, 
          fit=True, 
          target_col=target_col,
          feature_scaler=feature_scaler,
          target_scaler=target_scaler):
    """
    Scale the input features, using a separate scaler for the target.
    
    Parameters: 
    df (pd.DataFrame): Input dataframe
    fit (bool): Whether to fit the scaler to the data (only apply to training data)
    target_col (pd.Series): The column that is being predicted
    feature_scaler (StandardScaler): Scaler used for features
    target_scaler (StandardScaler): Scaler used for target
      
    Returns: 
    df_scaled (pd.DataFrame): Scaled dataframe   
    """    
    
    target = df[target_col].values.reshape(-1, 1)
    if fit:
        target_scaler.fit(target)
    target_scaled = target_scaler.transform(target)
    
    # Select all columns other than target to be features
    features = df.loc[:, df.columns != target_col].values
    
    if features.shape[1]:  # If there are any features
        if fit:
            feature_scaler.fit(features)
        features_scaled = feature_scaler.transform(features)
        
        # Combine target and features into one data frame
        df_scaled = pd.DataFrame(features_scaled)
        target_col_num = df.columns.get_loc(target_col)
        df_scaled.insert(target_col_num, target_col, target_scaled)
        df_scaled.columns = df.columns        
    
    else:  # If only target column (no additional features)
        df_scaled = pd.DataFrame(target_scaled, columns=df.columns)
      
    return df_scaled

def inverse_scale(data, target_scaler=target_scaler):
    """
    Transform the scaled values of the target back into their original form.
    The features are left alone, as we're assuming that the output of the model only includes the target.
    
    Parameters: 
    data (np.array): Input array
    target_scaler (StandardScaler): Scaler used for target
      
    Returns: 
    data_scaled (np.array): Scaled array   
    """    
    
    df = pd.DataFrame()
    data_scaled = np.empty([data.shape[1], data.shape[0]])
    for i in range(data.shape[1]):
        data_scaled[i] = target_scaler.inverse_transform(data[:,i])
    return data_scaled.transpose()

df_train_scaled=scale(df_train)
df_test_scaled=scale(df_test, False)
# Review scaled values

df_train_scaled.head()<jupyter_output><empty_output><jupyter_text>Create sequences of time series data<jupyter_code>def reframe(data, n_input_steps = n_input_steps, n_output_steps = n_output_steps, target_col = target_col):

    target_col_num = data.columns.get_loc(target_col)    
    
    # Iterate through data and create sequences of features and outputs
    df = pd.DataFrame(data)
    cols=list()
    for i in range(n_input_steps, 0, -1):
        cols.append(df.shift(i))
    for i in range(0, n_output_steps):
        cols.append(df.shift(-i))
        
    # Concatenate values and remove any missing values
    df = pd.concat(cols, axis=1)
    df.dropna(inplace=True)
    
    # Split the data into feature and target variables
    n_feature_cols = n_input_steps * n_features
    features = df.iloc[:,0:n_feature_cols]
    target_cols = [i for i in range(n_feature_cols + target_col_num, n_feature_cols + n_output_steps * n_features, n_features)]
    targets = df.iloc[:,target_cols]

    return (features, targets)

X_train_reframed, y_train_reframed = reframe(df_train_scaled)
X_test_reframed, y_test_reframed = reframe(df_test_scaled)<jupyter_output><empty_output><jupyter_text>Build a model and submit your training job to AI PlatformThe model you're building here trains pretty fast so you could train it in this notebook, but for more computationally expensive models, it's useful to train them in the Cloud. To use AI Platform Training, you'll package up your training code and submit a training job to the AI Platform Prediction service.In your training script, you'll also export your trained `SavedModel` to a Cloud Storage bucket.  Prepare test data<jupyter_code># Reshape test data to match model inputs and outputs

X_train = X_train_reframed.values.reshape(-1, n_input_steps, n_features)
X_test = X_test_reframed.values.reshape(-1, n_input_steps, n_features)
y_train = y_train_reframed.values.reshape(-1, n_output_steps)
y_test = y_test_reframed.values.reshape(-1, n_output_steps)
# Specify directories to be used later

TRAINER_DIR = 'trainer'
EXPORT_DIR = 'tf_export'
# Create trainer directory if it doesn't already exist

!mkdir $TRAINER_DIR
!touch $TRAINER_DIR/__init__.py
# Copy numpy arrays to npy files

np.save(TRAINER_DIR + '/x_train.npy', X_train)
np.save(TRAINER_DIR + '/x_test.npy', X_test)
np.save(TRAINER_DIR + '/y_train.npy', y_train)
np.save(TRAINER_DIR + '/y_test.npy', y_test)<jupyter_output><empty_output><jupyter_text>Prepare model code<jupyter_code># Write training code out to a file that will be submitted to the training job
# Note: f-strings are supported in Python 3.6 and above

model_template = f"""import argparse
import numpy as np
import os
import tempfile

from google.cloud import storage
from tensorflow import keras
from tensorflow.keras import Sequential
from tensorflow.keras.layers import Dense, LSTM
from tensorflow.keras.callbacks import EarlyStopping

n_features = {n_features} # Two features: y (previous values) and whether the date is a holiday
n_input_steps = {n_input_steps} # Lookback window
n_output_steps = {n_output_steps} # How many steps to predict forward

epochs = {epochs} # How many passes through the data (early-stopping will cause training to stop before this)
patience = {patience} # Terminate training after the validation loss does not decrease after this many epochs

def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--model-dir',
        default=os.getenv('AIP_MODEL_DIR'),
        help='URL to store the job output')
    args = parser.parse_args()
    print(args)
    return args

def main():
    args = get_args()
    print('args: ', args)
    model_dir = args.model_dir
    
    storage_client = storage.Client()
    bucket_name = model_dir.split('/')[2]
    
    bucket = storage_client.get_bucket(bucket_name)

    # Get the training data and convert back to np arrays
    local_data_dir = os.path.join(os.getcwd(), tempfile.gettempdir())
    
    data_files = ['x_train.npy', 'y_train.npy', 'x_test.npy', 'y_test.npy']
 
    for i in data_files:
        blob = storage.Blob('{TRAINER_DIR}/' + i, bucket)
        destination_file = local_data_dir + '/' + i
        open(destination_file, 'a').close()
        blob.download_to_filename(destination_file)

    X_train = np.load(local_data_dir + '/x_train.npy')
    y_train = np.load(local_data_dir + '/y_train.npy')
    X_test = np.load(local_data_dir + '/x_test.npy')
    y_test = np.load(local_data_dir + '/y_test.npy')
    
    # Build and train the model
    model = Sequential([
        LSTM({lstm_units}, input_shape=[n_input_steps, n_features], recurrent_activation=None),
        Dense(n_output_steps)])

    model.compile(optimizer='adam', loss='mae')

    early_stopping = EarlyStopping(monitor='val_loss', patience=patience)
    _ = model.fit(x=X_train, y=y_train, validation_data=(X_test, y_test), epochs=epochs, callbacks=[early_stopping])
    
    # Export the model
    export_path = os.path.join(model_dir, '{EXPORT_DIR}')
    model.save(export_path)
    
if __name__ == '__main__':
    main()
"""

with open(os.path.join(TRAINER_DIR, 'training_script.py'), 'w') as f:
    f.write(model_template.format(**globals()))
# Copy the model and training data files to a GCS bucket

!gsutil -m cp -r trainer $BUCKET_URI
# List the contents of the bucket to ensure they were copied properly

!gsutil ls $BUCKET_URI/$TRAINER_DIR<jupyter_output>gs://rthallam-demo-project/trainer/__init__.py
gs://rthallam-demo-project/trainer/training_script.py
gs://rthallam-demo-project/trainer/x_test.npy
gs://rthallam-demo-project/trainer/x_train.npy
gs://rthallam-demo-project/trainer/y_test.npy
gs://rthallam-demo-project/trainer/y_train.npy<jupyter_text>Launch a Training Job and Create a Model on AI Platform (Unified)<jupyter_code>job = aiplatform.CustomTrainingJob(
    display_name=f"train_forecast-lstm-nb-{uuid.uuid4()}",
    script_path="trainer/training_script.py",
    container_uri="gcr.io/cloud-aiplatform/training/tf-gpu.2-4:latest",
    requirements=["gcsfs==0.7.1"],
    model_serving_container_image_uri="gcr.io/cloud-aiplatform/prediction/tf2-gpu.2-4:latest",
)
aiplatform.start_run("custom-training-run-lstm")
params = {
    'input_steps': n_input_steps,
    'num_features': n_features,
    'output_steps': n_output_steps,
    'epochs': epochs,
    'optimizer': 'adam',
    'loss': 'mae'
    
}
aiplatform.log_params(params)

model = job.run(
    model_display_name="custom-train-forecasting-lstm",
    replica_count=1,
    machine_type="n1-standard-4",
    accelerator_type="NVIDIA_TESLA_K80",
    accelerator_count=1,
    args=[ f"--model-dir={BUCKET_URI}" ],
)
!gsutil ls gs://cloud-ai-platform-2f444b6a-a742-444b-b91a-c7519f51bd77/aiplatform-custom-training-2021-04-30-04:40:06.722/
# Set training job parameters

MODULE_NAME = TRAINER_DIR + '.model'
TRAIN_DIR = os.getcwd() + '/' + TRAINER_DIR
JOB_DIR = BUCKET_URI
# Submit the training job

!gcloud ai-platform jobs submit training $JOB_NAME \
        --scale-tier basic \
        --package-path $TRAIN_DIR \
        --module-name $MODULE_NAME \
        --job-dir $BUCKET_URI \
        --region $REGION \
        --runtime-version $RUNTIME_VERSION \
        --python-version $PYTHON_VERSION
# Check the job status

!gcloud ai-platform jobs describe $JOB_NAME<jupyter_output><empty_output><jupyter_text>Monitor output of your training jobFollow the instructions in the output of the gcloud command above to view the logs from your training job. You can also navigate to the [Jobs Section](https://console.cloud.google.com/ai-platform/jobs) of your Cloud Console to view logs.Once your training job completes successfully, it'll export your trained model as a TensorFlow `SavedModel` and write the output to a directory in your Cloud Storage bucket.<jupyter_code># Verify model was exported correctly

storage_client = storage.Client()
bucket = storage_client.get_bucket(BUCKET)
bucket_files = list(bucket.list_blobs(prefix=EXPORT_DIR + '/'))

# If you see a saved_model.pb and a variables/ and assets/ directory here, it means your model was exported correctly in your training job. Yay!

for file in bucket_files:
    print(file)<jupyter_output><empty_output><jupyter_text>Deploy a model version<jupyter_code># Create model if it doesn't already exist

!gcloud ai-platform models create $MODEL_NAME --regions $REGION
# Create the model version

export_path = BUCKET_URI +'/' + EXPORT_DIR
version = 'version_' + str(int(time.time()))

!gcloud ai-platform versions create $version \
  --model $MODEL_NAME \
  --origin $export_path \
  --runtime-version=$RUNTIME_VERSION \
  --framework $FRAMEWORK \
  --python-version=$PYTHON_VERSION<jupyter_output><empty_output><jupyter_text>Get predictions on deployed model<jupyter_code># Initialize client

endpoint = f'https://ml.googleapis.com'
client_options = ClientOptions(api_endpoint=endpoint)
service = googleapiclient.discovery.build('ml', 'v1', client_options=client_options, cache_discovery=False)
# Helper function to invoke the prediction service from
# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/master/ml_engine/online_prediction/predict.py

def predict_json(project, model, instances, version=None):
    """Send json data to a deployed model for prediction.
    Args:
        project (str): project where the AI Platform Model is deployed.
        model (str): model name.
        instances ([Mapping[str: Any]]): Keys should be the names of Tensors
            your deployed model expects as inputs. Values should be datatypes
            convertible to Tensors, or (potentially nested) lists of datatypes
            convertible to tensors.
        version: str, version of the model to target.
    Returns:
        Mapping[str: any]: dictionary of prediction results defined by the
            model.
    """

    name = 'projects/{}/models/{}'.format(project, model)

    if version is not None:
        name += '/versions/{}'.format(version)

    response = service.projects().predict(
        name=name,
        body={'instances': instances}
    ).execute()

    if 'error' in response:
        raise RuntimeError(response['error'])

    return response['predictions']
# Predict with the 1st element from the test set

prediction_json = {input_layer_name: X_test[0].tolist()}

pred_val = predict_json(PROJECT, MODEL_NAME, prediction_json)

pred_val
# Print prediction and compare to actual value

print('Predicted riders:', int(round(inverse_scale(np.array([pred_val[0]['dense'][0]]).reshape(1,1))[0][0])))
print('Actual riders:   ', int(round(inverse_scale(np.array([y_test[0]]))[0][0])))<jupyter_output><empty_output><jupyter_text>Cleanup<jupyter_code># Delete model version resource
!echo gcloud ai-platform versions delete {version} --model {MODEL_NAME} --quiet 

# Delete model resource
!echo gcloud ai-platform models delete {MODEL_NAME} --quiet<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09528003730376665, 'fraction_numerical': 0.008341536707942594, 'mean_word_length': 4.408237601569067, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 71, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '298871', 'license': 'Apache-2.0', 'n_tokens_mistral': 6240, 'n_tokens_neox': 6082, 'n_words': 1734, 'path': 'forecasting/03-cloud-training.ipynb', 'repo_name': 'RajeshThallam/cloud-aiplatform-demos', 'size': '19301'}