This story contains new, firsthand information uncovered by the writer.

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

In the ==first part of this tutorial(https://hackernoon.com/mlops-with-databricks-and-spark-part-1)== series, we took the first steps for building an end-to-end MLOps pipeline using Databricks and Spark, guided by Databricks' reference architecture. 

Here's a recap of the key steps we covered: Setting Up the Unity Catalog for Medallion Architecture: We organized our data into bronze, silver, and gold layers within the Unity Catalog, establishing a structured and efficient data management system. 

Ingesting Data into Unity Catalog: We demonstrated how to import raw data into the system, ensuring consistency and quality for subsequent processing stages. 

Training the Model: Utilizing Databricks, we trained a machine learning model tailored to our dataset, following best practices for scalable and effective model development. 

Hyperparameter Tuning with HyperOpt: To enhance model performance, we employed HyperOpt to automate the search for optimal hyperparameters, improving accuracy and efficiency. 

Experiment Tracking with Databricks MLflow: We utilized MLflow to log and monitor our experiments, maintaining a comprehensive record of model versions, metrics, and parameters for easy comparison and reproducibility.  

With these foundational steps completed, your model is now primed for deployment. 

In this second part, we'll focus on integrating two critical components into our system:1. 

Batch Inference: Implementing batch processing to generate predictions on large datasets, suitable for applications like bulk scoring and periodic reporting.2. 

Online Inference (Model Serving): Setting up real-time model serving to provide immediate predictions, essential for interactive applications and services.3. 

Model Monitoring: to ensure your deployed models maintain optimal performance and reliability over time.Let's get into it Model DeploymentThe departure point of the last blog was model evaluation. 

Now imagine we did the comparison and found that our model shows a higher performance compare to this production model. 

As we want (assume) to use the model in production, we want to take advantage of all the data that we have. 

The next step is to train and test the model using the full dataset. 

Then persist our model for later use by deploying it as our champion model. 

Since this is the final model that we want to use for inference, we use the Feature Engineering client to train the model. 

This way we are not only track the model lineage easier, but also offload the schema validation and feature transformation (if any) to the client.```pythonwith mlflow.start_run(run_name="ALS_best_model") as run:        als = ALS()         Now we set the parameters for the method        als.setMaxIter(MAX_ITER)        .setSeed(SEED)        .setRegParam(best_params"REG_PARAM")        .setUserCol(COL_USER)        .setItemCol(COL_ITEM)        .setRatingCol(COL_LABEL)        .setRank(best_params"RANK")        mlflow.log_param("MAX_ITER", MAX_ITER)        mlflow.log_param("RANK", best_params"RANK")        mlflow.log_param("REG_PARAM", best_params"REG_PARAM")         Create the model with these parameters.        

model = als.fit(df_full_data)        drop predictions where users and products from the test test and didn't make it into the training set. 

in this case, the prediction is NaN        model.setColdStartStrategy('drop')         predictions = model.transform(df_full_data)        signature = infer_signature(model_input = df_full_data, model_output = predictions.select(COL_LABEL))        log the model        mlflow.spark.log_model(model, model_name,                               sample_input=df_full_data.limit(3),                               signature = signature,                               conda_env=mlflow.spark.get_default_conda_env(),                               registered_model_name=f"{catalog_name}.{model_schema}.

{model_name}")        evaluator = RegressionEvaluator(predictionCol=COL_PRED, labelCol=COL_LABEL)        rmse = evaluator.setMetricName("rmse").evaluate(predictions)        mlflow.log_metric('rmse', rmse)```we can also use the ==Feature Store or Feature Engineering APIs(https://docs.databricks.com/en/machine-learning/feature-store/train-models-with-feature-store.html)== to train and log the models```pythonmodel_info = fe.log_model(model=model,                           artifact_path = model_name,                          flavor=mlflow.spark,                          training_set=fe_full_data,                          conda_env=mlflow.spark.get_default_conda_env(),                          registered_model_name= f"{catalog_name}.{model_schema}.

{model_name}"                        )```when we use the feature engineering API we can view the model’s lineage in Catalog Explorer data lineage in Dataticks Unity Catalog(https://cdn.hackernoon.com/images/a98nFEjuohShlIKwuarlXIkY1ID2-ay034p9.jpeg)Now lets update the model description and assign a Champion label to it.```pythonimport timefrom mlflow.tracking.client import MlflowClientfrom mlflow.entities.model_registry.model_version_status import ModelVersionStatusclient = MlflowClient()find the latest model versionmodel_name_path = f"{catalog_name}.{model_schema}.

{model_name}"model_version_infos = client.search_model_versions(f"name ='{model_name_path}'")new_model_version = max(int(model_version_info.version) for model_version_info in model_version_infos)add the model and model version descirptionclient.update_registered_model(  name=model_name_path,  description="collaborative filtering using Spark mllib ALS. 

This model use rating table")client.update_model_version(  name=model_name_path,  version=new_model_version,  description="this model is optimized Rank and REG_PARAM with Hyperopt and rmse as a loss function. 

trained on the full dataset") assign aliasclient.set_registered_model_alias(model_name_path, "Champion", new_model_version)```Now go ahead and check the schema that you registered the model. 

you should see all your updates as follows Model registry in Databricks Unity Catalog(https://cdn.hackernoon.com/images/a98nFEjuohShlIKwuarlXIkY1ID2-rg1342y.jpeg):::tipModel stages: If you use workspace for model registry you should stages to manage your models. 

Using aliases won’t work. Check out ==here(https://docs.databricks.com/en/mlflow/workspace-model-registry-example.html)== (https://docs.databricks.com/en/mlflow/workspace-model-registry-example.html)to see how it works::: Model Inference Batch ScoringNow imagine we want to use our model in production for inference. 

In this step we load the champion model and use it to generate 20 movie recommendations for each users.```pythonfrom mlflow.spark import load_model as spark_load_modelfrom mlflow.tracking.client import MlflowClientfrom create_training_set import split_data-- set UC as model registray mlflow.set_registry_uri("databricks-uc")-- initate mlflow clientclient = MlflowClient() -- read the config filewith open('config.json') as config_file:    config = json.load(config_file)catalog_name = config"catalog_name"gold_layer = config"gold_layer_name"silver_layer = config"silver_layer_name"user_item_table_name = config"user_item_table_name"ft_user_item_name = config"ft_user_item_name"model_name = config"model_name"model_schema = config"model_schema"-- create the model urimodel_path = f"{catalog_name}.{model_schema}.

{model_name}" --create the model_uri: there are two ways to do this 1: using the alias (we use this)model_version_uri = f"models:/{model_uri}@champion" 2: using model versionchampion_version = client.get_model_version_by_alias(model_uri, "champion")model_version_uri = f"models:/{model_uri}/{champion_version.version}" -- load the model pipline and exctract the modelmodel_pipeline = spark_load_model(model_version_uri)model = model_pipeline.stages0 -- batch scoring using the the modelfe_full_data, df_full_data, df_train, df_test = split_data()df_batch_input = df_full_data.drop("rating")df_scores = model.transform(df_batch_input) --- in case you used Feature Engineerin

In the ==first part of this tutorial(https://hackernoon.com/mlops-with-databricks-and-spark-part-1)== series, we took the first steps for building an end-to-end MLOps pipeline using Databricks and Spark, guided by Databricks' reference architecture. Here's a recap of the key steps we covered: Setting Up the Unity Catalog for Medallion Architecture: We organized our data into bronze, silver, and gold layers within the Unity Catalog, establishing a structured and efficient data management system. Ingesting Data into Unity Catalog: We demonstrated how to import raw data into the system, ensuring consistency and quality for subsequent processing stages. Training the Model: Utilizing Databricks, we trained a machine learning model tailored to our dataset, following best practices for scalable and effective model development. Hyperparameter Tuning with HyperOpt: To enhance model performance, we employed HyperOpt to automate the search for optimal hyperparameters, improving accuracy and efficiency. Experiment Tracking with Databricks MLflow: We utilized MLflow to log and monitor our experiments, maintaining a comprehensive record of model versions, metrics, and parameters for easy comparison and reproducibility.  With these foundational steps completed, your model is now primed for deployment. In this second part, we'll focus on integrating two critical components into our system:1. Batch Inference: Implementing batch processing to generate predictions on large datasets, suitable for applications like bulk scoring and periodic reporting.2. Online Inference (Model Serving): Setting up real-time model serving to provide immediate predictions, essential for interactive applications and services.3. Model Monitoring: to ensure your deployed models maintain optimal performance and reliability over time.Let's get into it Model DeploymentThe departure point of the last blog was model evaluation. Now imagine we did the comparison and found that our model shows a higher performance compare to this production model. As we want (assume) to use the model in production, we want to take advantage of all the data that we have. The next step is to train and test the model using the full dataset. Then persist our model for later use by deploying it as our champion model. Since this is the final model that we want to use for inference, we use the Feature Engineering client to train the model. This way we are not only track the model lineage easier, but also offload the schema validation and feature transformation (if any) to the client.```pythonwith mlflow.start_run(run_name="ALS_best_model") as run:        als = ALS()         Now we set the parameters for the method        als.setMaxIter(MAX_ITER)        .setSeed(SEED)        .setRegParam(best_params"REG_PARAM")        .setUserCol(COL_USER)        .setItemCol(COL_ITEM)        .setRatingCol(COL_LABEL)        .setRank(best_params"RANK")        mlflow.log_param("MAX_ITER", MAX_ITER)        mlflow.log_param("RANK", best_params"RANK")        mlflow.log_param("REG_PARAM", best_params"REG_PARAM")         Create the model with these parameters.        model = als.fit(df_full_data)        drop predictions where users and products from the test test and didn't make it into the training set. in this case, the prediction is NaN        model.setColdStartStrategy('drop')         predictions = model.transform(df_full_data)        signature = infer_signature(model_input = df_full_data, model_output = predictions.select(COL_LABEL))        log the model        mlflow.spark.log_model(model, model_name,                               sample_input=df_full_data.limit(3),                               signature = signature,                               conda_env=mlflow.spark.get_default_conda_env(),                               registered_model_name=f"{catalog_name}.{model_schema}.{model_name}")        evaluator = RegressionEvaluator(predictionCol=COL_PRED, labelCol=COL_LABEL)        rmse = evaluator.setMetricName("rmse").evaluate(predictions)        mlflow.log_metric('rmse', rmse)```we can also use the ==Feature Store or Feature Engineering APIs(https://docs.databricks.com/en/machine-learning/feature-store/train-models-with-feature-store.html)== to train and log the models```pythonmodel_info = fe.log_model(model=model,                           artifact_path = model_name,                          flavor=mlflow.spark,                          training_set=fe_full_data,                          conda_env=mlflow.spark.get_default_conda_env(),                          registered_model_name= f"{catalog_name}.{model_schema}.{model_name}"                        )```when we use the feature engineering API we can view the model’s lineage in Catalog Explorer data lineage in Dataticks Unity Catalog(https://cdn.hackernoon.com/images/a98nFEjuohShlIKwuarlXIkY1ID2-ay034p9.jpeg)Now lets update the model description and assign a Champion label to it.```pythonimport timefrom mlflow.tracking.client import MlflowClientfrom mlflow.entities.model_registry.model_version_status import ModelVersionStatusclient = MlflowClient()find the latest model versionmodel_name_path = f"{catalog_name}.{model_schema}.{model_name}"model_version_infos = client.search_model_versions(f"name ='{model_name_path}'")new_model_version = max(int(model_version_info.version) for model_version_info in model_version_infos)add the model and model version descirptionclient.update_registered_model(  name=model_name_path,  description="collaborative filtering using Spark mllib ALS. This model use rating table")client.update_model_version(  name=model_name_path,  version=new_model_version,  description="this model is optimized Rank and REG_PARAM with Hyperopt and rmse as a loss function. trained on the full dataset") assign aliasclient.set_registered_model_alias(model_name_path, "Champion", new_model_version)```Now go ahead and check the schema that you registered the model. you should see all your updates as follows Model registry in Databricks Unity Catalog(https://cdn.hackernoon.com/images/a98nFEjuohShlIKwuarlXIkY1ID2-rg1342y.jpeg):::tipModel stages: If you use workspace for model registry you should stages to manage your models. Using aliases won’t work. Check out ==here(https://docs.databricks.com/en/mlflow/workspace-model-registry-example.html)== (https://docs.databricks.com/en/mlflow/workspace-model-registry-example.html)to see how it works::: Model Inference Batch ScoringNow imagine we want to use our model in production for inference. In this step we load the champion model and use it to generate 20 movie recommendations for each users.```pythonfrom mlflow.spark import load_model as spark_load_modelfrom mlflow.tracking.client import MlflowClientfrom create_training_set import split_data-- set UC as model registray mlflow.set_registry_uri("databricks-uc")-- initate mlflow clientclient = MlflowClient() -- read the config filewith open('config.json') as config_file:    config = json.load(config_file)catalog_name = config"catalog_name"gold_layer = config"gold_layer_name"silver_layer = config"silver_layer_name"user_item_table_name = config"user_item_table_name"ft_user_item_name = config"ft_user_item_name"model_name = config"model_name"model_schema = config"model_schema"-- create the model urimodel_path = f"{catalog_name}.{model_schema}.{model_name}" --create the model_uri: there are two ways to do this 1: using the alias (we use this)model_version_uri = f"models:/{model_uri}@champion" 2: using model versionchampion_version = client.get_model_version_by_alias(model_uri, "champion")model_version_uri = f"models:/{model_uri}/{champion_version.version}" -- load the model pipline and exctract the modelmodel_pipeline = spark_load_model(model_version_uri)model = model_pipeline.stages0 -- batch scoring using the the modelfe_full_data, df_full_data, df_train, df_test = split_data()df_batch_input = df_full_data.drop("rating")df_scores = model.transform(df_batch_input) --- in case you used Feature Engineerin

Part 2 - MLOps Pipeline With Databricks and Spark - Depp

About Author

Topics

Around The Web

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps