# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python [default]
#     language: python
#     name: python3
# ---

# # Who Am I?
# ## <NAME>
#
# ![<NAME>](https://s3.amazonaws.com/fluxcapacitor.com/img/fregly-300x300.png)
#
# ![<NAME> Emoji](https://s3.amazonaws.com/fluxcapacitor.com/img/fregly-emoji-300x300.png)
# ## Research Scientist, Founder @ **[PipelineIO](http://pipeline.io)** 
#
# ![PipelineIO](http://pipeline.io/images/pipeline-io-logo-shadow-210x186.png)
#
# ## Video Series Author "High Performance Tensorflow in Production" @ [OReilly](http://oreilly.com) (Coming Soon)
#
# ![OReilly Media](https://s3.amazonaws.com/fluxcapacitor.com/img/oreilly-logo.png)
#
# ## Founder @ **[Advanced Spark and Tensorflow Meetup](http://http://www.meetup.com/Advanced-Spark-and-TensorFlow-Meetup/)**
#
# ![Advanced Spark and Tensorflow Meetup](https://s3.amazonaws.com/fluxcapacitor.com/img/advanced-spark-tensorflow-meetup-logo-green.png)
#
# ## **[Github Repo](https://github.com/fluxcapacitor/pipeline)**
# 	
# ![Github Repo](https://s3.amazonaws.com/fluxcapacitor.com/img/pipeline-io-github-1000-stars.png)
#
# ## **[DockerHub Repo](https://hub.docker.com/u/fluxcapacitor/)**
#
# ![DockerHub](https://s3.amazonaws.com/fluxcapacitor.com/img/dockerhub-logo.png)
#
# ## **[Slideshare](http://www.slideshare.net/cfregly)**
#
# ![Slideshare](http://advancedspark.com/img/slideshare.png)
#
# ## **[YouTube](https://www.youtube.com/playlist?list=PL7pBcJ870QHeNRBXdKirc4fdtbtbB5Xy-)**
#
# ![YouTube](http://advancedspark.com/img/youtube-300x134.png)

# # Who Was I?
# ## Software Engineer @ **Netflix**, **Databricks**, **IBM Spark Tech Center**
#
# ![Netflix](http://pipeline.io/images/netflixoss-logo-white-295x55.png) 
#
# ![Databricks](http://pipeline.io/images/databricks-logo-350x69.png) 
#
# ![IBM Spark Tech Center](https://s3.amazonaws.com/fluxcapacitor.com/img/ibm-spark-logo-197x148.png)

# # Types of Model Deployments
# ## KeyValue
# ie. Recommendations
#
# In-memory: Redis, Memcache
#
# On-disk:  Cassandra, RocksDB
#
# First-class Servable in [Tensorflow Serving](https://github.com/tensorflow/serving/tree/master/tensorflow_serving/servables)
#
# ## PMML
# It's Useful and [Well-Supported](https://github.com/jpmml)
#
# Apple, Cisco, [Airbnb](http://nerds.airbnb.com/architecting-machine-learning-system-risk/), HomeAway, etc
#
# Please Don't Re-build It - Reduce Your Technical Debt!
#
# ![Regression Model PMML](https://s3.amazonaws.com/fluxcapacitor.com/img/regression-model-pmml2.png)
#
# ## Native Code Generation (CPU and GPU)
# Hand-coded (Python + Pickling)
#
# Generate Java Code from PMML?
#
# ![Regression Model Java Codegen](https://s3.amazonaws.com/fluxcapacitor.com/img/regression-model-codegen2.png)
#
# ## Tensorflow Models
# [freeze_graph.py](https://github.com/tensorflow/tensorflow/blob/master/tensorflow/python/tools/freeze_graph.py): Combine Tensorflow Graph (Static) with Trained Weights (Checkpoints) into Single Deployable Model

# # Demos!!

# +
# You may need to Reconnect (more than Restart) the Kernel to pick up changes to these sett
import os

master = '--master spark://spark-master-2-1-0:7077'
conf = '--conf spark.cores.max=1 --conf spark.executor.memory=512m'
packages = '--packages com.amazonaws:aws-java-sdk:1.7.4,org.apache.hadoop:hadoop-aws:2.7.1'
jars = '--jars /root/lib/jpmml-sparkml-package-1.0-SNAPSHOT.jar'
py_files = '--py-files /root/lib/jpmml.py'

os.environ['PYSPARK_SUBMIT_ARGS'] = master \
  + ' ' + conf \
  + ' ' + packages \
  + ' ' + jars \
  + ' ' + py_files \
  + ' ' + 'pyspark-shell'

print(os.environ['PYSPARK_SUBMIT_ARGS'])
# -

# ## Deploy Spark ML Models

from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.feature import OneHotEncoder, StringIndexer
from pyspark.ml import Pipeline, PipelineModel
from pyspark.ml.regression import LinearRegression

# +
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
# -

# ### **Step 0**: Load Libraries and Data

# +
df = spark.read.format("csv") \
  .option("inferSchema", "true").option("header", "true") \
  .load("s3a://datapalooza/airbnb/airbnb.csv.bz2")

df.registerTempTable("df")

print(df.head())
# -

print(df.count())

# ### **Step 1**: Clean, Filter, and Summarize the Data

# +
df_filtered = df.filter("price >= 50 AND price <= 750 AND bathrooms > 0.0 AND bedrooms is not null")

df_filtered.registerTempTable("df_filtered")

df_final = spark.sql("""
    select
        id,
        city,
        case when state in('NY', 'CA', 'London', 'Berlin', 'TX' ,'IL', 'OR', 'DC', 'WA')
            then state
            else 'Other'
        end as state,
        space,
        cast(price as double) as price,
        cast(bathrooms as double) as bathrooms,
        cast(bedrooms as double) as bedrooms,
        room_type,
        host_is_super_host,
        cancellation_policy,
        cast(case when security_deposit is null
            then 0.0
            else security_deposit
        end as double) as security_deposit,
        price_per_bedroom,
        cast(case when number_of_reviews is null
            then 0.0
            else number_of_reviews
        end as double) as number_of_reviews,
        cast(case when extra_people is null
            then 0.0
            else extra_people
        end as double) as extra_people,
        instant_bookable,
        cast(case when cleaning_fee is null
            then 0.0
            else cleaning_fee
        end as double) as cleaning_fee,
        cast(case when review_scores_rating is null
            then 80.0
            else review_scores_rating
        end as double) as review_scores_rating,
        cast(case when square_feet is not null and square_feet > 100
            then square_feet
            when (square_feet is null or square_feet <=100) and (bedrooms is null or bedrooms = 0)
            then 350.0
            else 380 * bedrooms
        end as double) as square_feet
    from df_filtered
""").persist()

df_final.registerTempTable("df_final")

df_final.select("square_feet", "price", "bedrooms", "bathrooms", "cleaning_fee").describe().show()
# -

print(df_final.count())

print(df_final.schema)

# +
# Most popular cities

spark.sql("""
    select 
        state,
        count(*) as ct,
        avg(price) as avg_price,
        max(price) as max_price
    from df_final
    group by state
    order by count(*) desc
""").show()

# +
# Most expensive popular cities

spark.sql("""
    select 
        city,
        count(*) as ct,
        avg(price) as avg_price,
        max(price) as max_price
    from df_final
    group by city
    order by avg(price) desc
""").filter("ct > 25").show()
# -

# ### **Step 2**: Define Continous and Categorical Features

# +
continuous_features = ["bathrooms", \
                       "bedrooms", \
                       "security_deposit", \
                       "cleaning_fee", \
                       "extra_people", \
                       "number_of_reviews", \
                       "square_feet", \
                       "review_scores_rating"]

categorical_features = ["room_type", \
                        "host_is_super_host", \
                        "cancellation_policy", \
                        "instant_bookable", \
                        "state"]
# -

# ### **Step 3**: Split Data into Training and Validation 

[training_dataset, validation_dataset] = df_final.randomSplit([0.8, 0.2])

# ### **Step 4**: Continous Feature Pipeline

# +
continuous_feature_assembler = VectorAssembler(inputCols=continuous_features, outputCol="unscaled_continuous_features")

continuous_feature_scaler = StandardScaler(inputCol="unscaled_continuous_features", outputCol="scaled_continuous_features", \
                                           withStd=True, withMean=False)
# -

# ### **Step 5**: Categorical Feature Pipeline
#

# +
categorical_feature_indexers = [StringIndexer(inputCol=x, \
                                              outputCol="{}_index".format(x)) \
                                for x in categorical_features]

categorical_feature_one_hot_encoders = [OneHotEncoder(inputCol=x.getOutputCol(), \
                                                      outputCol="oh_encoder_{}".format(x.getOutputCol() )) \
                                        for x in categorical_feature_indexers]
# -

# ### **Step 6**: Assemble our Features and Feature Pipeline

# +
feature_cols_lr = [x.getOutputCol() \
                   for x in categorical_feature_one_hot_encoders]
feature_cols_lr.append("scaled_continuous_features")

feature_assembler_lr = VectorAssembler(inputCols=feature_cols_lr, \
                                       outputCol="features_lr")
# -

# ### **Step 7**: Train a Linear Regression Model

# +
linear_regression = LinearRegression(featuresCol="features_lr", \
                                     labelCol="price", \
                                     predictionCol="price_prediction", \
                                     maxIter=10, \
                                     regParam=0.3, \
                                     elasticNetParam=0.8)

estimators_lr = \
  [continuous_feature_assembler, continuous_feature_scaler] \
  + categorical_feature_indexers + categorical_feature_one_hot_encoders \
  + [feature_assembler_lr] + [linear_regression]

pipeline = Pipeline(stages=estimators_lr)

pipeline_model = pipeline.fit(training_dataset)

print(pipeline_model)
# -

# ### **Step 8**:  Serialize PipelineModel
#

# +
from jpmml import toPMMLBytes

model_bytes = toPMMLBytes(spark, training_dataset, pipeline_model)

print(model_bytes.decode("utf-8"))
# -

# ### Step 9: Push Model to Live, Running Spark ML Model Server (Mutable)

# +
import urllib.request

namespace = 'default'
model_name = 'airbnb'
version = '1'
update_url = 'http://prediction-pmml-aws.demo.pipeline.io/update-pmml-model/%s/%s/%s' % (namespace, model_name, version)

update_headers = {}
update_headers['Content-type'] = 'application/xml'

req = urllib.request.Request(update_url, \
                             headers=update_headers, \
                             data=model_bytes)

resp = urllib.request.urlopen(req)

print(resp.status) # Should return Http Status 200 
# -

# ### Step 10:  Evalute Model

# +
import urllib.parse
import json

namespace = 'default'
model_name = 'airbnb'
version = '1'
evaluate_url = 'http://prediction-pmml-aws.demo.pipeline.io/evaluate-pmml-model/%s/%s/%s' % (namespace, model_name, version)

evaluate_headers = {}
evaluate_headers['Content-type'] = 'application/json'

input_params = '{"bathrooms":5.0, \
                 "bedrooms":4.0, \
                 "security_deposit":175.00, \
                 "cleaning_fee":25.0, \
                 "extra_people":1.0, \
                 "number_of_reviews": 2.0, \
                 "square_feet": 250.0, \
                 "review_scores_rating": 2.0, \
                 "room_type": "Entire home/apt", \
                 "host_is_super_host": "0.0", \
                 "cancellation_policy": "flexible", \
                 "instant_bookable": "1.0", \
                 "state": "CA"}' 
encoded_input_params = input_params.encode('utf-8')

req = urllib.request.Request(evaluate_url, \
                             headers=evaluate_headers, \
                             data=encoded_input_params)

resp = urllib.request.urlopen(req)

print(resp.read())
# -

# # Bonus Demos!

# ## Deploy Java-based Model

# ### Create Java-based Model

# +
from urllib import request

sourceBytes = '                                                      \n\
  private String str;                                                \n\
                                                                     \n\
  public void initialize(Map<String, Object> args) {                 \n\
  }                                                                  \n\
                                                                     \n\
  public Object predict(Map<String, Object> inputs) {                \n\
      String id = (String)inputs.get("id");                          \n\
                                                                     \n\
      return id.equals("21619");                                     \n\
  }                                                                  \n\
'.encode('utf-8')
# -

# ### Deploy Java-based Model

# +
from urllib import request

namespace = 'default'
model_name = 'java_equals'
version = '1'

update_url = 'http://prediction-java-aws.demo.pipeline.io/update-java/%s/%s/%s' % (namespace, model_name, version)

update_headers = {}
update_headers['Content-type'] = 'text/plain'

req = request.Request("%s" % update_url, headers=update_headers, data=sourceBytes)
resp = request.urlopen(req)

generated_code = resp.read()
print(generated_code.decode('utf-8'))
# -

# ### Evaluate Java-based Model

# +
from urllib import request

namespace = 'default'
model_name = 'java_equals'
version = '1'

evaluate_url = 'http://prediction-java-aws.demo.pipeline.io/evaluate-java/%s/%s/%s' % (namespace, model_name, version)

evaluate_headers = {}
evaluate_headers['Content-type'] = 'application/json'
input_params = '{"id":"21618"}' 
encoded_input_params = input_params.encode('utf-8')

req = request.Request(evaluate_url, headers=evaluate_headers, data=encoded_input_params)
resp = request.urlopen(req)

print(resp.read()) # Should return false 

# +
from urllib import request

namespace = 'default'
model_name = 'java_equals'
version = '1'
evaluate_url = 'http://prediction-java-aws.demo.pipeline.io/evaluate-java/%s/%s/%s' % (namespace, model_name, version)

evaluate_headers = {}
evaluate_headers['Content-type'] = 'application/json'
input_params = '{"id":"21619"}' 
encoded_input_params = input_params.encode('utf-8')

req = request.Request(evaluate_url, headers=evaluate_headers, data=encoded_input_params)
resp = request.urlopen(req)

print(resp.read()) # Should return true
# -

# ## Deploy Scikit-Learn Model

# !pip install sklearn_pandas
# !pip install git+https://github.com/jpmml/sklearn2pmml.git

# ### Create Scikit-Learn Model

# +
import pandas as pd
import numpy as np
import urllib.request
import urllib.parse
import json

from sklearn.datasets import load_diabetes,load_iris
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error as mse, r2_score
from sklearn2pmml import PMMLPipeline
from sklearn.tree import DecisionTreeClassifier
from sklearn2pmml import sklearn2pmml

iris = load_iris()
iris_df = pd.DataFrame(iris.data,columns=iris.feature_names)
iris_df['Species'] = iris.target
iris_pipeline = PMMLPipeline([
    ("classifier", DecisionTreeClassifier())
])
iris_pipeline.fit(iris_df[iris_df.columns.difference(["Species"])], iris_df["Species"])
# -

# ### Serialize Scikit-Learn Model

sklearn2pmml(iris_pipeline, "DecisionTreeIris.pmml", with_repr = True)
model_bytes = bytearray(open('DecisionTreeIris.pmml', 'rb').read())

# ### Deploy Scikit-Learn Model

# +
import urllib.request
import urllib.parse

namespace = 'default'
model_name = 'iris'
version = '1'

update_url = 'http://prediction-pmml-aws.demo.pipeline.io/update-pmml-model/%s/%s/%s' % (namespace, model_name, version)

update_headers = {}
update_headers["Content-type"] = "application/xml"

req = urllib.request.Request(update_url, headers=update_headers, data=model_bytes)

resp = urllib.request.urlopen(req)
print(resp.status)

# +
namespace = 'default'
model_name = 'iris'
version = '1'

evaluate_url = 'http://prediction-pmml-aws.demo.pipeline.io/evaluate-pmml-model/%s/%s/%s' % (namespace, model_name, version)
evaluate_headers = {}
evaluate_headers['Content-type'] = 'application/json'

input_params = iris_df.ix[0,:-1].to_json()
encoded_input_params = input_params.encode('utf-8')

req = urllib.request.Request(evaluate_url, headers=evaluate_headers, data=encoded_input_params)
resp = urllib.request.urlopen(req)

print(resp.read())
# -

# ## Monitoring Your Models

# ### Netflix Microservices [Dashboard](http://hystrix.demo.pipeline.io/hystrix-dashboard/monitor/monitor.html?streams=%5B%7B%22name%22%3A%22Predictions%20-%20AWS%22%2C%22stream%22%3A%22http%3A%2F%2Fturbine-aws.demo.pipeline.io%2Fturbine.stream%22%2C%22auth%22%3A%22%22%2C%22delay%22%3A%22%22%7D%2C%7B%22name%22%3A%22Predictions%20-%20GCP%22%2C%22stream%22%3A%22http%3A%2F%2Fturbine-gcp.demo.pipeline.io%2Fturbine.stream%22%2C%22auth%22%3A%22%22%2C%22delay%22%3A%22%22%7D%5D) (Hystrix)

# +
from IPython.core.display import display, HTML
display(HTML("<style>.container { width:100% !important; }</style>"))

from IPython.display import clear_output, Image, display, HTML

html = '<iframe width=1200px height=500px src="http://hystrix.demo.pipeline.io/hystrix-dashboard/monitor/monitor.html?streams=%5B%7B%22name%22%3A%22Predictions%20-%20AWS%22%2C%22stream%22%3A%22http%3A%2F%2Fturbine-aws.demo.pipeline.io%2Fturbine.stream%22%2C%22auth%22%3A%22%22%2C%22delay%22%3A%22%22%7D%2C%7B%22name%22%3A%22Predictions%20-%20GCP%22%2C%22stream%22%3A%22http%3A%2F%2Fturbine-gcp.demo.pipeline.io%2Fturbine.stream%22%2C%22auth%22%3A%22%22%2C%22delay%22%3A%22%22%7D%5D">'
display(HTML(html))
# -

# ### Grafana + Prometheus [Dashboard](http://grafana.demo.pipeline.io)

# +
from IPython.core.display import display, HTML
display(HTML("<style>.container { width:100% !important; }</style>"))

from IPython.display import clear_output, Image, display, HTML

html = '<iframe width=1200px height=500px src="http://grafana.demo.pipeline.io">'
display(HTML(html))
# -

# # Load-Test Your Model Servers

# ### Run JMeter Tests from Local Laptop (Limited by Laptop Performance)

# ### Run Headless JMeter Tests from Training Clusters in Cloud

# Spark ML - Airbnb
# !kubectl create --context=awsdemo -f /root/pipeline/loadtest.ml/loadtest-aws-airbnb-rc.yaml

# Codegen - Java - Simple
# !kubectl create --context=awsdemo -f /root/pipeline/loadtest.ml/loadtest-aws-equals-rc.yaml

# Tensorflow AI - Tensorflow Serving - Simple 
# !kubectl create --context=awsdemo -f /root/pipeline/loadtest.ml/loadtest-aws-minimal-rc.yaml

# ### End Load Tests

# !kubectl delete --context=awsdemo rc loadtest-aws-airbnb
# !kubectl delete --context=awsdemo rc loadtest-aws-equals
# !kubectl delete --context=awsdemo rc loadtest-aws-minimal

# ## Rolling Deploy

# !kubectl rolling-update prediction-tensorflow --context=awsdemo --image-pull-policy=Always --image=fluxcapacitor/prediction-tensorflow

# # PipelineIO Premium Edition
#
# ![PipelineIO](http://pipeline.io/images/pipeline-io-logo-shadow-210x186.png)

# ## A/B and Multi-armed Bandit Testing
#
# ![A/B Testing](http://pipeline.io/images/ab-testing-170x100.png)
# ![Multi-armed Bandit](http://pipeline.io/images/multi-armed-bandit-101x100.png)

# ## Continuous, Hybrid-Cloud Deployments
# ![AWS](http://pipeline.io/images/aws-logo-185x73.png)
# ![Google](http://pipeline.io/images/gce-logo-190x90.png)
# ![Azure](http://pipeline.io/images/azure-logo-200x103.png)
# ![Kubernetes](http://pipeline.io/images/kubernetes-logo-200x171.png)
# ![Docker](http://pipeline.io/images/docker-logo-150x126.png)

# ## Online Model Training and Deploying
#
# ![Kafka](http://pipeline.io/images/kafka-logo-wide-219x98.png)

# ## GPU-based Deployments
#
# ![CUDA](http://pipeline.io/images/nvidia-cuda-338x181.png)
#



StarcoderdataJupyterScriptsDedupFiltered