Developing HANA ML models with SAP Databricks

[nidhi_sawhney]

Introduction

SAP HANA provides a rich set of Machine Learning capabilities natively which can be used via SQL or python interface. For an introduction to these capabilities you can refer to HANA Machine Learning and Developing Regression Models with the Python Machine Learning Client for SAP HANA learning journey and this excellent blog post from @YannickSchaper.

In this blogpost I will walkthrough the capabilities that enhance the power of hana-ml with the model tracking capabilities provided by  mlflow . The python package hana-ml has supported the tracking and usability of trained ml models via mlflow which are covered extensively in these 2-part blogposts tracking-hana-machine-learning-experiments-with-mlflow-a-conceptual-guide from @stojanm and @martinboeckling . In this post I will focus on the Databricks managed mlflow as it greatly eases the use of mlflow without having to setup the mlflow server. These capabilities are available both in SAP Databricks from SAP Business Data Cloud(BDC) and Enterprise Databricks for customers who connect Databricks to BDC via bdc-connect. For this blogpost I will be using SAP Databricks provisioned with SAP Business Data Cloud.

With the launch of SAP Business Data Cloud, developers have a much more streamlined access to AI/ML capabilities both from SAP and Databricks. This applies to data available via the Unity Catalog or accessible via the SQL access. Here I will focus on the Notebook capabilities and Training and Inference on datasets in the HANA Cloud layer accessed via SQL and utilize the compute of HANA Cloud.

 

The datasets in HANA Cloud could be data persisted in HANA or remotely available via federation from HDLFS or BDC Data Products installed to embedded HANA Cloud from Datasphere.

Connect to ML datasets on HANA Cloud

To connect to data on the HANA Cloud, be it the embedded HANA Cloud of SAP Datasphere or a stand-alone HANA Cloud, one needs the 4 parameters which provide the url, port(443), username and password.

Prerequisites

In addition the HANA Cloud or Datasphere instance needs to have the Databricks IP to the Allow-list to enable connection.

The database user needs to have the following privileges which are provided by the HANA Cloud or Datasphere administrator 

1. AFL__SYS_AFL_AFLPAL_EXECUTE_WITH_GRANT_OPTION
2. AFL__SYS_AFL_APL_AREA_EXECUTE
3. AFLPM_CREATOR_ERASER_EXECUTE

For Datasphere these privileges are enabled when the administrator creates the database user with OpenSQL access and Enables APL and PAL

 

 

 

 

 

 

 

 

Connect to HANA from Databricks

Here is a code snippet to connect to the HANA Cloud SQL layer for data access using Databricks secret. For this you need to create the secrets needed for HANA Cloud connectivity like snippet below

from databricks.sdk import WorkspaceClient
w = WorkspaceClient() 
scope = "<scope-name>"
w.secrets.create_scope(scope)
url =  "<hana-url>"
port = 443
user = "<hana-db-user>"
password = "<hana-db-password>"
w.secrets.put_secret(scope,"hana_url",string_value =url)
w.secrets.put_secret(scope,"hana_port",string_value =port)
w.secrets.put_secret(scope,"hana_user",string_value =user)
w.secrets.put_secret(scope,"hana_password",string_value = password)

create_secrets

import os
import hana_ml
from hana_ml import dataframe
import mlflow

print("hana_ml version:", hana_ml.__version__)
print("mlflow version:", mlflow.__version__)
scope = "<scope_name>"
os.environ['HANA_ADDRESS'] = dbutils.secrets.get(scope=scope, key="hana_url")
os.environ['HANA_PORT']    = dbutils.secrets.get(scope=scope, key="hana_port")
os.environ['HANA_UNAME']   = dbutils.secrets.get(scope=scope, key="hana_user")
os.environ['HANA_PASS']    = dbutils.secrets.get(scope=scope, key="hana_password")
import hana_ml.dataframe as dataframe
cc = dataframe.ConnectionContext(
    address=os.environ['HANA_ADDRESS'],
    port=os.environ['HANA_PORT'],
    user=os.environ['HANA_UNAME'],
    password=os.environ['HANA_PASS']
)
if cc.connection.isconnected(): 
    print(f'User {os.environ["HANA_UNAME"]} connected to HANA successfully')
    print(f"HANA Version: {cc.hana_version()}")

Otherwise you can also connect via the python-dotenv, especially if you are developing locally.

 

Develop the ML model with mlflow

Here I will use a sample dataset provided by hana-ml package to make it easier to test. This would be replaced by the appropriate dataset the user wants to use for training the ML model.

from hana_ml.algorithms.pal.utility import DataSets
# Load Dataset
bike_dataset = DataSets.load_bike_data(cc)#This creates the correspoding table on HANA Cloud 
# number of rows and number of columns
print("Shape of datset: {}".format(bike_dataset.shape))
# columns
print(bike_dataset.columns)
# types of each column
print(bike_dataset.dtypes())

# print the first 3 rows of dataset
print(bike_dataset.head(3).collect())

#Split the dataset into train & test
# Add a ID column for AutomaticRegression, the last column is the label
bike_dataset = bike_dataset.add_id('ID', ref_col='days_since_2011')

# Split the dataset into training and test dataset
cols = bike_dataset.columns
cols.remove('cnt')
bike_data = bike_dataset[cols + ['cnt']]
bike_train = bike_data.filter('ID <= 600')
bike_test = bike_data.filter('ID > 600')
print(bike_train.head(3).collect())
print(bike_test.head(3).collect())

We used a basic splitting methodology above, hana-ml provides splitting capabilities via hana_ml.algorithms.pal.partition.train_test_val_split to assist in this process.

Now that we have  a training and test dataset, we can start the training process and use mlflow to track the results in Databricks experiments via the code below

mlflow.set_tracking_uri("databricks")
experiment_path = '<experiment_path>' 
mlflow.set_experiment(experiment_path)
# Here we are using AutomaticRegression to show the metrics automatically created and tracked via mlflow
from hana_ml.algorithms.pal.auto_ml import AutomaticClassification, AutomaticRegression
auto_r = AutomaticRegression(generations=2,
                             population_size=15,
                             offspring_size=5)

# enable_workload_class if you have workload_classes defined on HANA Cloud instance, here we disable it but in productive scenarios you would have it enabled
#auto_r.enable_workload_class(workload_class_name="PAL_AUTOML_WORKLOAD")
auto_r.disable_workload_class_check()
try:
    with mlflow.start_run(run_name="hana-ml-autoreg-bike") as run:
        auto_r.enable_mlflow_autologging(is_exported=True)
        auto_r.fit(bike_train, key="ID")
        runid = run.info.run_id
except Exception as e:
    raise e

The enable_mlflow_autologging function above enables the creation of key model metrics automatically, in this case suitable for regression without any additional effort from the user. These metrics would differ based on the algorithm. The user can easily log additional parameters, metrics, artifacts as desired and supported my mlfow.

When the above code is run we get the experiments logged with default metrics that hana-ml model logged automatically via mlflow for example R2, RMSE etc below.

One can then compare different runs and track model progression as parameters are changed.

For inferencing on data the hana-ml model can be loaded to HANA Cloud via the code below, the run_id is the run from the Databricks Experiments that you would like to use for inference and can be obtained from the overview of the Experiment

from hana_ml.model_storage import ModelStorage
bikemodel = ModelStorage.load_mlflow_model(connection_context=cc, model_uri='runs:/{}/model'.format(runid))

#Get the info for the loaded model
bikemodel.mlflow_model_info

#Use the trained model for prediction on test or new dataset
res = bikemodel.predict(bike_test.deselect('cnt') , key="ID")
print(res.collect())
bike_test.deselect('cnt').save("INFERENCE_BIKE_DATA_TBL") #Saving this here for using later via the Serving Endpoint

Serve the ML model for inferencing

The hana-ml model can be served on BTP if desirable by exporting the ml model or store and reload the model for inference from HANA Cloud hana_ml.model_storage , in this case the HANA Cloud instance would need to be same for training and inferencing.

Alternatively, it can be served on Databricks via the Serving endpoint, I describe these below.

Databricks does not natively support the serving for the hana-ml model. However, this can be achieved via the mlflow.pyfunc functionality to provide custom models. I will be using the "model from code" method as it has advantages over the legacy methods and recommended going forward. This requires passing the custom handler as separate code. For this we write the python file which handles the desired input to the serving endpoint. In my example, I want the user to pass in the name of table which exists in HANA Cloud(in our example we saved it as  

INFERENCE_BIKE_DATA_TBL

and has the data that needs to be inferenced. The user sends the name of the table to the inference endpoint. The code can be modified to have the input say as a payload to the inference end-point, in that case the custom handler function(hana_ml_pyfunc_model) would then need to persist the payload as a HANA table so the hana-ml predict can be called on it.

Create the custom handler for hana-ml

# Save as script: hana_ml_pyfunc_model.py
# %%writefile "./hana_ml_pyfunc_model.py"
import mlflow
from mlflow import pyfunc
from mlflow.models import set_model
import hana_ml
from hana_ml import dataframe
from hana_ml.model_storage import ModelStorage
import os


class hana_ml_pyfunc_model(pyfunc.PythonModel):

    def connectToHANA(self, context):
        try:
            url =  os.getenv('hana_url') 
            port = os.getenv('hana_port')
            user = os.getenv('hana_user')
            passwd = os.getenv('hana_password')
            connection_context = dataframe.ConnectionContext(url, port, user, passwd)
            return connection_context
        except Exception as e:
            print(f"Exception occurred: {e}")
            raise e
            return "Exception:{e}", e
    @mlflow.trace
    def load_context(self, context):
        try: 
            with mlflow.start_span("load_context"):
                self.model = context.artifacts["model"]
                self.connection_context = self.connectToHANA(context)
                print("HANA_ML_MODEL loaded in load_context")
        except Exception as e:
            print(f"Exception occurred: {e}")
            raise Exception(f"Loading the context failed due to {e}")
    @mlflow.trace
    def predict(self, context, model_input):
        table_name = None
        try: 
            if self.connection_context.connection.isconnected() == False:
                with mlflow.start_span("connect_to_HANA"):
                    self.connection_context = self.connectToHANA(context)
                    if self.connection_context.connection.isconnected():
                        print("HANA Connection Successful")
                    else:
                        raise Exception("HANA Connection Failed")
       
            with mlflow.start_span("load_model"):
                hana_model = ModelStorage.load_mlflow_model(connection_context=self.connection_context, model_uri=self.model,use_temporary_table=False, force=True)
           
                print("HANA_ML_MODEL loaded in predict")
                print("model_input", model_input)
                table_name = str(model_input["INFERENCE_TABLE_NAME"][0]) 
                print("Table Name:", table_name)
            with mlflow.start_span("hana_ml_predict"):
                df = self.connection_context.table(table_name)
                if df.count() > 0:
                  
                    print(f"Running HANA ML inference on {table_name} with {df.count()} records")
                    prediction = hana_model.predict(df, key = "ID").collect()
                    print("Prediction completed")
                else:
                    raise Exception(f"HANA Inference Table {table_name} is empty")
        
            return prediction
        except Exception as e:
            print(f"Exception occurred: {e}")
            raise f"Exception:{e}"
        
set_model(hana_ml_pyfunc_model())

 

Log the custom pyfunc model

Then we log the above pyfunc model which can be registered to enable the creation of a Serving endpoint on Databricks

#Create the signature for the model input and output. In this example:
# the input is the name of an existing table in HANA which has the data that needs to be inferenced 
# the output is the "cnt" counts for the bike data and the associated score
import mlflow
from mlflow.models import ModelSignature, infer_signature
from mlflow.types.schema import Schema, ColSpec
signature = ModelSignature(inputs = Schema([ColSpec("string", "INFERENCE_TABLE_NAME")]))
signature.outputs = Schema([ColSpec("integer", "ID"), ColSpec("double", "SCORES")])


mlflow.set_tracking_uri("databricks")
runid="<run_id>" #runid from the training phase which is the chosen champion model to be served
model_uri='runs:/{}/model'.format(runid)
experiment_name = '<experiment_name>' 
mlflow.set_experiment(experiment_name)

model_file = "hana_ml_pyfunc_model.py" #This is the file that is written in step above and handles the calll to hana_ml for predict on the user-provided inference table

with mlflow.start_run() as run:
    mlflow.pyfunc.log_model(
            artifact_path="model",
            python_model=model_file,
            artifacts={"model": model_uri},
            pip_requirements=["hana-ml","ipython"],
            signature = signature,
            input_example={"INFERENCE_TABLE_NAME" : "INFERENCE_BIKE_DATA_TBL"},
    )
# Register the model #Need to register the model to enable it to be served on Databricks
model_uri = f"runs:/{run.info.run_id}/model"
registered_model_name = "<your_model_name>"
mlflow.register_model(model_uri=model_uri, name=registered_model_name)

 

Test the custom pyfunc model

To test the logged model in step above you can call the following code

## Code to test the model logged as custom pyfunc model which can be registered and deployed for serving
logged_model = 'runs:/run.info.run_id/model' #run from the pyfunc model logging
dataset = {"inputs": {"INFERENCE_TABLE_NAME" : "INFERENCE_BIKE_DATA_TBL"}}
loaded_model = mlflow.pyfunc.load_model(logged_model)
loaded_model.predict(dataset["inputs"])

Additionally the model endpoint can also be tested by using package uv with code below

run_id = run.info.run_id #run from the pyfunc model logging
model_uri = f"runs:/{run_id}/model"
dataset = {"inputs": {"INFERENCE_TABLE_NAME" : "INFERENCE_BIKE_DATA_TBL"}}
input_data = dataset
mlflow.models.predict(
    model_uri=model_uri,
    input_data=dataset["inputs"],
    env_manager="uv",
)

 

Create the Serving Endpoint

Now we have a registered model that can be deployed for serving. To do this I show the steps here to do it via the Databricks Serving UI  via the serving endpoint, it can also be done with code via Rest API or sdks. Go to Serving and Create a new Serving endpoint. Choose the registered_model_name from step above and add the environment variables for the HANA Cloud connection so the serving code can connect to HANA and call the model inference on user provided table name

 

 

 

 

 

 

 

 

 

Test the Serving Endpoint

The deployment as usual takes some minutes. Once the serving endpoint is in ready state, it can be tested with the usual ways when pressing Use 

Here is a sample screenshot for testing it in the Browser

Here is the corresponding code to test via Python

import os,json,requests
os.environ['DATABRICKS_TOKEN'] = "<Developer_Token>" #Token obtained by following https://docs.databricks.com/aws/en/dev-tools/auth/pat

def score_model(dataset):
    url = '<serving_url>'
    headers = {'Authorization': f'Bearer {os.environ.get("DATABRICKS_TOKEN")}', 'Content-Type': 'application/json'}
    data_json = json.dumps(dataset, allow_nan=True)
    response = requests.request(method='POST', headers=headers, url=url, data=data_json)
    if response.status_code != 200:
        raise Exception(f'Request failed with status {response.status_code}, {response.text}')
    return response.json()

dataset = {'inputs': {'INFERENCE_TABLE_NAME' : "<hana_cloud_table_name_for_inference>"}}
res = score_model(dataset)
print(res)

 

Endpoint Consumption

The serving endpoint created above can be used in applications via rest API calls. In production the API would need to be secured via OAuth authenication . Here is a blogpost from @Ian_Henry  describing how such an endpoint can be triggered from SAP Analytics Cloud for example.

Conclusion

In this blogpost we show the power combination of using SAP HANA Cloud and model experiment tracking & serving capabilities from SAP Databricks via managed mlflow. This is suitable for usecases where data already resides in the HANA layer and the performance benefits from running hana-ml in data accessible via HANA Cloud in-memory is desirable while benefitting from model development support provided by SAP Databricks. 

The code for the above is available here on SAP-samples/hana-ml-samples.