<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> Machine Learning Model training and serving </h1>
The training architecture involves collecting data from the two sources, mashing them up with Dataflow and saving the results in BigQuery. That data is used for ML training in Cloud ML.  Then, the trained model is used to orchestrate windmills.
<img src="training_arch.png" />
Step1:  <h2> Step 1
Step3:  <h2> 2. Preprocessing </h2>
Align the radar and windmill data temporally and compute moving averages.
Step4:  <h2> Local preprocessing, training and prediction </h2>
Before we setup a large-scale ML pipeline, it is best practice to try out the code on a small dataset. This can be done on the machine on which you are running the Datalab notebook.
Our pipeline consists of the following steps
Step5:  <h3> Training </h3>
Step6:  <h3> Prediction </h3>
Step7:  <h2> Cloud deploy model and predict </h2>
Now that we have a working model, we can turn the model loose and train on the full dataset. The model can then be deployed, which essentially puts it on the Cloud and attaches a REST API to it so that we can invoke the trained model from the windmills.
Step8:  After the model is deployed, we will want to test it.  An easy way to test a deployed ML model is to write out a file with some hypothetical inputs and then use gcloud to invoke the REST API.
<h2> Invoking REST API </h2>
Here is how the windmills invoke our model.  Note that invoking a deployed ML model is just a REST API call. The windmill owners don't know anything about what ML model we are running and this way, those details are all hidden away. The input variables may be scaled during preprocessing, but again the client doesn't know any of that.
<img src="serving_arch.png"/>
This preserves maximum flexbility.
<ASSISTANT_TASK:>
Python Code:
%projects set ml-autoawesome
import os
PROJECT = 'ml-autoawesome'    # CHANGE THIS
BUCKET = 'ml-autoawesome-cmle'  # CHANGE THIS
REGION = 'us-central1' # CHANGE THIS
os.environ['PROJECT'] = PROJECT # for bash
os.environ['BUCKET'] = BUCKET # for bash
os.environ['REGION'] = REGION # for bash
%bash
echo "project=$PROJECT"
echo "bucket=$BUCKET"
echo "region=$REGION"
gcloud config set project $PROJECT
gcloud config set compute/region $REGION
#gcloud beta ml init-project -q
Explanation: <h1> Machine Learning Model training and serving </h1>
The training architecture involves collecting data from the two sources, mashing them up with Dataflow and saving the results in BigQuery. That data is used for ML training in Cloud ML.  Then, the trained model is used to orchestrate windmills.
<img src="training_arch.png" />
End of explanation
import tensorflow as tf
import google.datalab.ml as ml
import apache_beam as beam
from tensorflow.python.lib.io import file_io
import json
import shutil
INDIR = '.'
OUTDIR = '.'
os.environ['OUTDIR'] = OUTDIR
Explanation: <h2> Step 1:  Import Python packages that we need </h2>
End of explanation
def make_preprocessing_fn():
  # stop-gap ...
  def _scalar_to_vector(scalar):
    # FeatureColumns expect shape (batch_size, 1), not just (batch_size)
    return api.map(lambda x: tf.expand_dims(x, -1), scalar)
  
  def preprocessing_fn(inputs):
    result = {col: _scalar_to_vector(inputs[col]) for col in CSV_COLUMNS}
    for name in SCALE_COLUMNS:
      result[name] = _scalar_to_vector(mappers.scale_to_0_1(inputs[name]))
    return result
  return preprocessing_fn
def make_input_schema(mode):
  input_schema = {}
  if mode != tf.contrib.learn.ModeKeys.INFER:
      input_schema[LABEL_COLUMN] = tf.FixedLenFeature(shape=[], dtype=tf.float32, default_value=0.0)
  for name in ['dayofweek', 'key']:
      input_schema[name] = tf.FixedLenFeature(shape=[], dtype=tf.string, default_value='null')
  for name in ['hourofday']:
      input_schema[name] = tf.FixedLenFeature(shape=[], dtype=tf.int64, default_value=0)
  for name in SCALE_COLUMNS:
      input_schema[name] = tf.FixedLenFeature(shape=[], dtype=tf.float32, default_value=0.0)
      
  input_schema = dataset_schema.from_feature_spec(input_schema)
  return input_schema
def make_coder(schema, mode):
  import copy
  column_names = copy.deepcopy(CSV_COLUMNS)
  if mode == tf.contrib.learn.ModeKeys.INFER:
    column_names.pop(LABEL_COLUMN)
  coder = coders.CsvCoder(column_names, schema)
  return coder
def preprocess_all(pipeline, training_data, eval_data, predict_data, output_dir, mode=tf.contrib.learn.ModeKeys.TRAIN):
  path_constants = PathConstants()
  work_dir = os.path.join(output_dir, path_constants.TEMP_DIR)
  
  # create schema
  input_schema = make_input_schema(mode)
  # coder
  coder = make_coder(input_schema, mode)
  # 3) Read from text using the coder.
  train_data = (
      pipeline
      | 'ReadTrainingData' >> beam.io.ReadFromText(training_data)
      | 'ParseTrainingCsv' >> beam.Map(coder.decode))
  evaluate_data = (
      pipeline
      | 'ReadEvalData' >> beam.io.ReadFromText(eval_data)
      | 'ParseEvalCsv' >> beam.Map(coder.decode))
  # metadata
  input_metadata = dataset_metadata.DatasetMetadata(schema=input_schema)
  _ = (input_metadata
       | 'WriteInputMetadata' >> io.WriteMetadata(
           os.path.join(output_dir, path_constants.RAW_METADATA_DIR),
           pipeline=pipeline))
  preprocessing_fn = make_preprocessing_fn()
  (train_dataset, train_metadata), transform_fn = (
      (train_data, input_metadata)
      | 'AnalyzeAndTransform' >> tft.AnalyzeAndTransformDataset(
          preprocessing_fn, work_dir))
  # WriteTransformFn writes transform_fn and metadata to fixed subdirectories
  # of output_dir, which are given by path_constants.TRANSFORM_FN_DIR and
  # path_constants.TRANSFORMED_METADATA_DIR.
  transform_fn_is_written = (transform_fn | io.WriteTransformFn(output_dir))
  (evaluate_dataset, evaluate_metadata) = (
      ((evaluate_data, input_metadata), transform_fn)
      | 'TransformEval' >> tft.TransformDataset())
  train_coder = coders.ExampleProtoCoder(train_metadata.schema)
  _ = (train_dataset
       | 'SerializeTrainExamples' >> beam.Map(train_coder.encode)
       | 'WriteTraining'
       >> beam.io.WriteToTFRecord(
           os.path.join(output_dir,
                        path_constants.TRANSFORMED_TRAIN_DATA_FILE_PREFIX),
           file_name_suffix='.tfrecord.gz'))
  evaluate_coder = coders.ExampleProtoCoder(evaluate_metadata.schema)
  _ = (evaluate_dataset
       | 'SerializeEvalExamples' >> beam.Map(evaluate_coder.encode)
       | 'WriteEval'
       >> beam.io.WriteToTFRecord(
           os.path.join(output_dir,
                        path_constants.TRANSFORMED_EVAL_DATA_FILE_PREFIX),
           file_name_suffix='.tfrecord.gz'))
  if predict_data:
    predict_mode = tf.contrib.learn.ModeKeys.INFER
    predict_schema = make_input_schema(mode=predict_mode)
    tsv_coder = make_coder(predict_schema, mode=predict_mode)
    predict_coder = coders.ExampleProtoCoder(predict_schema)
    _ = (pipeline
         | 'ReadPredictData' >> beam.io.ReadFromText(predict_data,
                                                     coder=tsv_coder)
         # TODO(b/35194257) Obviate the need for this explicit serialization.
         | 'EncodePredictData' >> beam.Map(predict_coder.encode)
         | 'WritePredictData' >> beam.io.WriteToTFRecord(
             os.path.join(output_dir,
                          path_constants.TRANSFORMED_PREDICT_DATA_FILE_PREFIX),
             file_name_suffix='.tfrecord.gz'))
  # Workaround b/35366670, to ensure that training and eval don't start before
  # the transform_fn is written.
  train_dataset |= beam.Map(
      lambda x, y: x, y=beam.pvalue.AsSingleton(transform_fn_is_written))
  evaluate_dataset |= beam.Map(
      lambda x, y: x, y=beam.pvalue.AsSingleton(transform_fn_is_written))
  return transform_fn, train_dataset, evaluate_dataset
p = beam.Pipeline()
output_dataset = 'windmills_control'
transform_fn, train_dataset, eval_dataset = preprocess_all(
      p, train_data_paths, eval_data_paths, predict_data_paths, output_dataset)
p.run()
import pandas as pd
import numpy as np
import datalab.bigquery as bq
# get data from BigQuery
query=
SELECT
  speed, weight, angular_momentum, wind_dir, moisture_content, radar_reflectivity, radar_cap,
  radar_distance_to_nearest_cloud, radar_x, radar_y
FROM
  windmill_control
df = bq.Query(query).to_dataframe()
# Add a unique key (needed for batch prediction)
df['key'] = 1000 + df.index.values
# Use Pandas to create 90% training & 10% evaluation
df = df.reindex(np.random.permutation(df.index))
trainsize = (len(df)*9)/10
df_train = df.head(trainsize)
df_eval = df.tail(len(df) - trainsize)
df_train.to_csv('cleanedup-train.csv', header=False, index_label=False, index=False)
df_eval.to_csv('cleanedup-eval.csv', header=False, index_label=False, index=False)
df.head()
%bash
ls -lrt cleanedup*
Explanation: <h2> 2. Preprocessing </h2>
Align the radar and windmill data temporally and compute moving averages.
End of explanation
!rm -rf ml_preproc ml_trained
train_bq = ml.BigQueryDataSet(
  table_pattern=('cleanedup-train*'),
  schema_file=os.path.join('ml.json'))
sd.local_preprocess(
  dataset=train_bq,
  output_dir=os.path.join(OUTDIR, 'ml_preproc'),
)
file_io.write_string_to_file(os.path.join(OUTDIR, 'ml_preproc/transforms.json'),
                             json.dumps(transforms, indent=2))
!cat $OUTDIR/ml_preproc/num*json
Explanation: <h2> Local preprocessing, training and prediction </h2>
Before we setup a large-scale ML pipeline, it is best practice to try out the code on a small dataset. This can be done on the machine on which you are running the Datalab notebook.
Our pipeline consists of the following steps:
<ol>
<li> Preprocessing: this goes through the full dataset and computes min/max/mean of the input columns and target. These are useful as (a) defaults in case some input is missing in production  (b) to scale the inputs, because some ML optimizers work better on scaled inputs.
<li> Training: this consists of adjusting weights on the model to reduce the error on the training dataset.
<li> Evaluation: how well does this model do on the validation dataset? In the packaged solution, we can pass in the validation dataset and the evaluation will be carried out periodically during training itself. We don't need to do this separately.
<li> Prediction: try out the trained model on some inputs to get an idea of what the model does in some hypothetical situation.
</ol>
<p/>
<h3> Preprocessing </h3>
End of explanation
eval_bq = ml.BigQueryDataSet(
  file_pattern=('cleanedup-eval*'),
  schema_file=os.path.join(OUTDIR, 'ml.json'))
shutil.rmtree(os.path.join(OUTDIR, 'ml_trained'), ignore_errors=True)
sd.local_train(
  train_dataset=train_bq,
  eval_dataset=eval_bq,
  preprocess_output_dir=os.path.join(OUTDIR, 'ml_preproc'),
  transforms=os.path.join(OUTDIR, 'ml_preproc/transforms.json'),
  output_dir=os.path.join(OUTDIR, 'ml_trained'),
  model_type='dnn_regression',
  max_steps=2500,
  layer_sizes=[1024]*4
)
%bash
ls $OUTDIR/ml_trained
Explanation: <h3> Training </h3>
End of explanation
import pandas as pd
df = pd.read_csv('{}/batch_predict/predictions-00000-of-00001.csv'.format(OUTDIR), names=('key','true_cost','predicted_cost'))
df['true_cost'] = df['true_cost'] * 20000
df['predicted_cost'] = df['predicted_cost'] * 20000
df.head()
import seaborn as sns
sns.jointplot(x='true_cost', y="predicted_cost", data=df, kind='hex');
Explanation: <h3> Prediction </h3>
End of explanation
%bash
MODEL_NAME="windmill"
MODEL_VERSION="v3"
MODEL_LOCATION="/content/autoawesome/notebooks/ml_trained/model"
echo "Deleting and deploying $MODEL_NAME $MODEL_VERSION from $MODEL_LOCATION ... this will take a few minutes"
gcloud beta ml versions delete ${MODEL_VERSION} --model ${MODEL_NAME}
gcloud beta ml models delete ${MODEL_NAME}
gcloud beta ml models create ${MODEL_NAME} --regions $REGION
gcloud beta ml versions create ${MODEL_VERSION} --model ${MODEL_NAME} --staging-bucket gs://${BUCKET} --origin ${MODEL_LOCATION}
Explanation: <h2> Cloud deploy model and predict </h2>
Now that we have a working model, we can turn the model loose and train on the full dataset. The model can then be deployed, which essentially puts it on the Cloud and attaches a REST API to it so that we can invoke the trained model from the windmills.
End of explanation
from googleapiclient import discovery
from oauth2client.client import GoogleCredentials
import json
#import google.cloud.ml.features as features
#from google.cloud.ml import session_bundle
credentials = GoogleCredentials.get_application_default()
api = discovery.build('ml', 'v1beta1', credentials=credentials,
            discoveryServiceUrl='https://storage.googleapis.com/cloud-ml/discovery/ml_v1beta1_discovery.json')
request_data = {'instances':
  [
     # redacted to protect privacy of our windmill owners
  ]
}
parent = 'projects/%s/models/%s/versions/%s' % (PROJECT, 'windmills', 'v3')
response = api.projects().predict(body=request_data, name=parent).execute()
print "response={0}".format(response)
Explanation: After the model is deployed, we will want to test it.  An easy way to test a deployed ML model is to write out a file with some hypothetical inputs and then use gcloud to invoke the REST API.
<h2> Invoking REST API </h2>
Here is how the windmills invoke our model.  Note that invoking a deployed ML model is just a REST API call. The windmill owners don't know anything about what ML model we are running and this way, those details are all hidden away. The input variables may be scaled during preprocessing, but again the client doesn't know any of that.
<img src="serving_arch.png"/>
This preserves maximum flexbility.
End of explanation
<END_TASK>