<jupyter_start><jupyter_text>Read a csv from object storageconvert to spark dataframe<jupyter_code>df = spark.read.format("csv").option('header', 'true').load("oci://hosted-ds-datasets@bigdatadatasciencelarge/titanic/titanic.csv")
df.show(5)
# Column types: 
df.dtypes<jupyter_output><empty_output><jupyter_text>Data preparation and feature engineering<jupyter_code># Typecast columns. Survived is the target variable: 
dataset = df.select(col('Survived').cast('float'),
                         col('Pclass').cast('float'),
                         col('Sex'),
                         col('Age').cast('float'),
                         col('Fare').cast('float')
                        )

dataset.show()
# Index categorical columns with StringIndexer

dataset = StringIndexer(inputCol='Sex', 
                        outputCol='Gender', 
                        handleInvalid='keep').fit(dataset).transform(dataset)

# drop columns 
dataset = dataset.drop('Sex')
dataset = dataset.drop('Embarked')

dataset.show()<jupyter_output><empty_output><jupyter_text>Build a simple Pipeline Object<jupyter_code># Input features: 
required_features = ['Pclass',
                    'Age',
                    'Fare',
                    'Gender']

# 80/20 data split 
(training_data, test_data) = dataset.randomSplit([0.8,0.2])

assembler = VectorAssembler(inputCols=required_features, outputCol='features')
rf = RandomForestClassifier(labelCol='Survived', 
                            featuresCol='features',
                            maxDepth=5)

pipeline = Pipeline(stages=[assembler, rf])
pipeline_model = pipeline.fit(training_data)
results = pipeline_model.transform(dataset)
results.limit(10).show()<jupyter_output><empty_output><jupyter_text>Saving the Model Object to disk<jupyter_code>artifact_dir = "./model-artifact"

if not os.path.exists(artifact_dir):
    os.makedirs(artifact_dir)
pipeline_model.write().overwrite().save(f"{artifact_dir}/my-model")
#pipeline_model.write().overwrite().save("oci://<my-bucket>@<my-namespace/my-model")<jupyter_output><empty_output><jupyter_text>Preparing a model artifact with ADS<jupyter_code>ads_artifact = prepare_generic_model(artifact_dir, 
                                     force_overwrite=True, 
                                     function_artifacts=False, 
                                     data_science_env=True)<jupyter_output><empty_output><jupyter_text>Overwriting score.py to support spark models Added a bunch of print statements for logging purposes.<jupyter_code>%%writefile {artifact_dir}/score.py

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline, PipelineModel
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import RandomForestClassifier
from pyspark.sql.functions import col


sc = SparkSession \
    .builder \
    .appName('Titanic2') \
    .getOrCreate()

"""
   Inference script. This script is used for prediction by scoring server when schema is known.
"""


def load_model():
    """
    Loads model from the serialized format

    Returns
    -------
    model:  a model instance on which predict API can be invoked
    """
    print('loading model')
    # use this path for model deployment: 
    pm2 = PipelineModel.load('/home/datascience/model-server/app/deployed_model/my-model')
    # use this path for testing locally: 
    #pm2 = PipelineModel.load('/home/datascience/sparkml/model-artifact/my-model/')
 
    print('done reading model')

    print(pm2.__class__)
    return pm2


def predict(data, model=load_model()) -> dict:
    """
    Returns prediction given the model and data to predict

    Parameters
    ----------
    model: Model instance returned by load_model API
    data: Data format as expected by the predict API of the core estimator. For eg. in case of sckit models it could be numpy array/List of list/Panda DataFrame

    Returns
    -------
    predictions: Output from scoring server
        Format: { 'prediction': output from `model.predict` method }

    """
    print('before reading data')
    tmp = sc.read.json(sc.sparkContext.parallelize([data]))
    print(tmp.show())
    print(tmp.printSchema())
    print('after reading data')
    tmp = tmp.select(col('Pclass').cast('float'),
                      col('Gender').cast('double'),
                      col('Age').cast('float'),
                      col('Fare').cast('float')
                     )
    print('after selecting subset of columns and typecasting')   
    results = model.transform(tmp)
    print('after prediction') 
    print(results.show())
    res = results.toPandas()['prediction']
    print("prediction = ", res)
    return { 'prediction': str(list(res)) }<jupyter_output><empty_output><jupyter_text>Testing the model locally<jupyter_code>import sys 
sys.path.insert(0, artifact_dir)

from score import load_model, predict
_ = load_model()
# Create a sample payload as a json string
test = test_data.toPandas()
testjson = test[:1].to_json(orient='records',lines=True)
testjson
res = predict(testjson, _)
# Saving the artifact to the Catalog 
ads_artifact.save(display_name="my-spark-model", 
                  description="pipeline object",
                  project_id=os.environ['PROJECT_OCID'],
                  compartment_id=os.environ['NB_SESSION_COMPARTMENT_OCID'])<jupyter_output><empty_output><jupyter_text>Deploy the model through the console   Invoke the model<jupyter_code>import requests
import oci
from oci.signer import Signer

# User principal auth: 
#config = oci.config.from_file("~/.oci/config") # replace with the location of your oci config file
#auth = Signer(
#  tenancy=config['tenancy'],
#  user=config['user'],
#  fingerprint=config['fingerprint'],
#  private_key_file_location=config['key_file'],
#  pass_phrase=config['pass_phrase'])

# Resource principal auth:
auth = oci.auth.signers.get_resource_principals_signer()

# replace with your own endpoint: 
endpoint = ''

requests.post(endpoint, json=testjson, auth=auth).json()
%%time 
requests.post(endpoint, json=testjson, auth=auth).json()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11972409262604697, 'fraction_numerical': 0.003120381014944983, 'mean_word_length': 3.3781452192667145, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '485992', 'license': 'Apache-2.0', 'n_tokens_mistral': 1856, 'n_tokens_neox': 1784, 'n_words': 390, 'path': 'sparkml/spark-e2e.ipynb', 'repo_name': 'jrgauthier01/oci-ds-samples', 'size': '6089'}