<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <h1> 2d. Distributed training and monitoring </h1>
In this notebook, we refactor to call train_and_evaluate instead of hand-coding our ML pipeline. This allows us to carry out evaluation as part of our training loop instead of as a separate step. It also adds in failure-handling that is necessary for distributed training capabilities.
Step1:  <h2> Input </h2>
Read data created in Lab1a, but this time make it more general, so that we are reading in batches.  Instead of using Pandas, we will use add a filename queue to the TensorFlow graph.
Step2:  <h2> Create features out of input data </h2>
For now, pass these through.  (same as previous lab)
Step3:  <h2> Serving input function </h2>
Defines the expected shape of the JSON feed that the modelwill receive once deployed behind a REST API in production.
Step4:  <h2> tf.estimator.train_and_evaluate </h2>
Step5:  <h2>Run training</h2>
<ASSISTANT_TASK:>
Python Code:
!sudo chown -R jupyter:jupyter /home/jupyter/training-data-analyst
# Ensure the right version of Tensorflow is installed.
!pip freeze | grep tensorflow==2.5
from google.cloud import bigquery
import tensorflow as tf
import numpy as np
import shutil
print(tf.__version__)
Explanation: <h1> 2d. Distributed training and monitoring </h1>
In this notebook, we refactor to call train_and_evaluate instead of hand-coding our ML pipeline. This allows us to carry out evaluation as part of our training loop instead of as a separate step. It also adds in failure-handling that is necessary for distributed training capabilities.
End of explanation
CSV_COLUMNS = ['fare_amount', 'pickuplon','pickuplat','dropofflon','dropofflat','passengers', 'key']
LABEL_COLUMN = 'fare_amount'
DEFAULTS = [[0.0], [-74.0], [40.0], [-74.0], [40.7], [1.0], ['nokey']]
def read_dataset(filename, mode, batch_size = 512):
  def decode_csv(value_column):
    columns = tf.compat.v1.decode_csv(value_column, record_defaults = DEFAULTS)
    features = dict(zip(CSV_COLUMNS, columns))
    label = features.pop(LABEL_COLUMN)
    return features, label
  # Create list of file names that match "glob" pattern (i.e. data_file_*.csv)
  filenames_dataset = tf.data.Dataset.list_files(filename)
  # Read lines from text files
  textlines_dataset = filenames_dataset.flat_map(tf.data.TextLineDataset)
  # Parse text lines as comma-separated values (CSV)
  dataset = textlines_dataset.map(decode_csv)
  # Note:
  # use tf.data.Dataset.flat_map to apply one to many transformations (here: filename -> text lines)
  # use tf.data.Dataset.map      to apply one to one  transformations (here: text line -> feature list)
  if mode == tf.estimator.ModeKeys.TRAIN:
      num_epochs = None # indefinitely
      dataset = dataset.shuffle(buffer_size = 10 * batch_size)
  else:
      num_epochs = 1 # end-of-input after this
  dataset = dataset.repeat(num_epochs).batch(batch_size)
  return dataset
Explanation: <h2> Input </h2>
Read data created in Lab1a, but this time make it more general, so that we are reading in batches.  Instead of using Pandas, we will use add a filename queue to the TensorFlow graph.
End of explanation
INPUT_COLUMNS = [
    tf.feature_column.numeric_column('pickuplon'),
    tf.feature_column.numeric_column('pickuplat'),
    tf.feature_column.numeric_column('dropofflat'),
    tf.feature_column.numeric_column('dropofflon'),
    tf.feature_column.numeric_column('passengers'),
]
def add_more_features(feats):
  # Nothing to add (yet!)
  return feats
feature_cols = add_more_features(INPUT_COLUMNS)
Explanation: <h2> Create features out of input data </h2>
For now, pass these through.  (same as previous lab)
End of explanation
## TODO: Create serving input function
def serving_input_fn():
  #ADD CODE HERE
  return tf.estimator.export.ServingInputReceiver(features, json_feature_placeholders)
Explanation: <h2> Serving input function </h2>
Defines the expected shape of the JSON feed that the modelwill receive once deployed behind a REST API in production.
End of explanation
## TODO: Create train and evaluate function using tf.estimator
def train_and_evaluate(output_dir, num_train_steps):
  #ADD CODE HERE
  tf.estimator.train_and_evaluate(estimator, train_spec, eval_spec)
Explanation: <h2> tf.estimator.train_and_evaluate </h2>
End of explanation
OUTDIR = './taxi_trained'
shutil.rmtree(OUTDIR, ignore_errors = True) # start fresh each time
tf.compat.v1.summary.FileWriterCache.clear()
train_and_evaluate(OUTDIR, num_train_steps = 2000)
Explanation: <h2>Run training</h2>
End of explanation
<END_TASK>