Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import os , shutil , json
import pyspark
from azureml.core.model import Model
from pyspark.ml import PipelineModel
def init():
global model
global spark
#Creating an spark session
spark = pyspark.sql.SparkSession.builder.appName("tweets_decision_tree").getOrCreate()
model_name = "tweets_decision_tree"
#Loading the model
#My model is stored in Azure Machine Learning Services. If not your case, replace accordingly
model_path = Model.get_model_path(model_name)
model_unpacked = "./" + model_name
#Unpacking archive
shutil.unpack_archive(model_path, model_unpacked)
#Creating the PipelineModel object from path
model = PipelineModel.load(model_unpacked)
# when our server gets new request for classification/prediction/scoring, it calls the model run functionally with the raw_data
def run(raw_data):
# validate that the model is up
if isinstance(trainedModel, Exception):
#Loading rutine failed to load the model
return json.dumps({{"trainedModel":str(trainedModel)}})
try:
#Converting raw data into Dataframe (Spark)
input_df = spark.createDataFrame(input_data)
# Score using spark mllib decision tree pipeline - compute prediction
result = model.transform(input_df).collect()
# you can return any data type as long as it is JSON-serializable
return result
except Exception as e:
print('[ERR] Exception happened: ' + str(e))
result = 'Input ' + str(input_data) + '. Exception was: ' + str(e)
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment