<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 Datasets.
Step2:  <h2> Create features out of input data </h2>
For now, pass these through.  (same as previous lab)
Step3:  <h2> train_and_evaluate </h2>
<ASSISTANT_TASK:>
Python Code:
# Ensure the right version of Tensorflow is installed.
!pip freeze | grep tensorflow==2.6
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 _input_fn():
    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 files that match pattern
    file_list = tf.compat.v1.gfile.Glob(filename)
    # Create dataset from file list
    dataset = tf.compat.v1.data.TextLineDataset(file_list).map(decode_csv)
    
    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.make_one_shot_iterator().get_next()
  return _input_fn
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 Datasets.
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
def serving_input_fn():
  feature_placeholders = {
    'pickuplon' : tf.compat.v1.placeholder(tf.float32, [None]),
    'pickuplat' : tf.compat.v1.placeholder(tf.float32, [None]),
    'dropofflat' : tf.compat.v1.placeholder(tf.float32, [None]),
    'dropofflon' : tf.compat.v1.placeholder(tf.float32, [None]),
    'passengers' : tf.compat.v1.placeholder(tf.float32, [None]),
  }
  features = {
      key: tf.expand_dims(tensor, -1)
      for key, tensor in feature_placeholders.items()
  }
  return tf.estimator.export.ServingInputReceiver(features, feature_placeholders)
def train_and_evaluate(output_dir, num_train_steps):
  estimator = tf.estimator.LinearRegressor(
                       model_dir = output_dir,
                       feature_columns = feature_cols)
  train_spec=tf.estimator.TrainSpec(
                       input_fn = read_dataset('./taxi-train.csv', mode = tf.estimator.ModeKeys.TRAIN),
                       max_steps = num_train_steps)
  exporter = tf.estimator.LatestExporter('exporter', serving_input_fn)
  eval_spec=tf.estimator.EvalSpec(
                       input_fn = read_dataset('./taxi-valid.csv', mode = tf.estimator.ModeKeys.EVAL),
                       steps = None,
                       start_delay_secs = 1, # start evaluating after N seconds
                       throttle_secs = 10,  # evaluate every N seconds
                       exporters = exporter)
  tf.estimator.train_and_evaluate(estimator, train_spec, eval_spec)
# Run training    
OUTDIR = 'taxi_trained'
shutil.rmtree(OUTDIR, ignore_errors = True) # start fresh each time
train_and_evaluate(OUTDIR, num_train_steps = 5000)
Explanation: <h2> train_and_evaluate </h2>
End of explanation
<END_TASK>