How to save your fitted transformer into blob, so your prediction pipeline can use it in AML Service?

user_7757104 0
Viewed 1056

I am building a data transformation and training pipeline on Azure Machine Leaning Service. I'd like to save my fitted transformer (e.g. tf-idf) to the blob, so my prediction pipeline can later access it. 

transformed_data = PipelineData("transformed_data", 
                               datastore = default_datastore,
                               output_path_on_compute="my_project/tfidf")

step_tfidf = PythonScriptStep(name = "tfidf_step",
                              script_name = "transform.py",
                              arguments = ['--input_data', blob_train_data, 
                                           '--output_folder', transformed_data],
                              inputs = [blob_train_data],
                              outputs = [transformed_data],
                              compute_target = aml_compute,
                              source_directory = project_folder,
                              runconfig = run_config,
                              allow_reuse = False)

The above code saves the transformer to a current run's folder, which is dynamically generated during each run.

I want to save the transformer to a fixed location on blob, so I can access it later, when calling a prediction pipeline.

I tried to use an instance of DataReference class as PythonScriptStep output, but it results in an error: ValueError: Unexpected output type: <class 'azureml.data.data_reference.DataReference'>

It's because PythonScriptStep only accepts PipelineData or OutputPortBinding objects as outputs.

How could I save my fitted transformer so it's later accessible by any aribitraly process (e.g. my prediction pipeline)?

azure-machine-learning-service python
3 Answers

This is likely not flexible enough for your needs (also, I haven't tested this yet), but if you are using scikit-learn one possibility is to include the tf-idf/transformation step into a scikit-learn Pipeline object and register that into your workspace.

Your training script would thus contain:

pipeline = Pipeline([
    ('vectorizer', TfidfVectorizer(stop_words = list(text.ENGLISH_STOP_WORDS))),
    ('classifier', SGDClassifier()
])

pipeline.fit(train[label].values, train[pred_label].values)

# Serialize the pipeline
joblib.dump(value=pipeline, filename='outputs/model.pkl')

and your experiment submission script would contain

run = exp.submit(src)
run.wait_for_completion(show_output = True)
model = run.register_model(model_name='my_pipeline', model_path='outputs/model.pkl')

Then, you could use the registered "model" and deploy it as a service as explained in the documentation, by loading it into a scoring script via

model_path = Model.get_model_path('my_pipeline')
# deserialize the model file back into a sklearn model
model = joblib.load(model_path)

However this would bake the transformation in your pipeline, and thus would not be as modular as you ask...

user_42404130

Another solution is to pass DataReference as an input to your PythonScriptStep.

Then inside transform.py you're able to read this DataReference as a command line argument.

You can parse it and use it just as any regular path to save your vectorizer to.

E.g. you can:

step_tfidf = PythonScriptStep(name = "tfidf_step",
                              script_name = "transform.py",
                              arguments = ['--input_data', blob_train_data, 
                                           '--output_folder', transformed_data,
                                           '--transformer_path', trained_transformer_path],
                              inputs = [blob_train_data, trained_transformer_path],
                              outputs = [transformed_data],
                              compute_target = aml_compute,
                              source_directory = project_folder,
                              runconfig = run_config,
                              allow_reuse = False)

Then inside your script (transform.py in the example above) you can e.g.:

import argparse
import joblib as jbl
import os

from sklearn.feature_extraction.text import TfidfVectorizer

parser = argparse.ArgumentParser()
parser.add_argument('--transformer_path', dest="transformer_path", required=True)
args = parser.parse_args()

tfidf = ### HERE CREATE AND TRAIN YOUR VECTORIZER ###

vect_filename = os.path.join(args.transformer_path, 'my_vectorizer.jbl')

EXTRA: The third way would be to just register the vectorizer as another model in your workspace. You can then use it exactly as any other registered model. (Though this option does not involve explicit writing to blob - as specified in the question above)

user_77571040

Another option will be to use DataTransferStep and use it to copy the output to a "known location." This notebook has examples of using DataTransferStep to copy data from and to various supported datastores.

from azureml.data.data_reference import DataReference
from azureml.exceptions import ComputeTargetException
from azureml.core.compute import ComputeTarget, DataFactoryCompute
from azureml.pipeline.steps import DataTransferStep

blob_datastore = Datastore.get(ws, "workspaceblobstore")

blob_data_ref = DataReference(
    datastore=blob_datastore,
    data_reference_name="knownloaction",
    path_on_datastore="knownloaction")

data_factory_name = 'adftest'

def get_or_create_data_factory(workspace, factory_name):
    try:
        return DataFactoryCompute(workspace, factory_name)
    except ComputeTargetException as e:
        if 'ComputeTargetNotFound' in e.message:
            print('Data factory not found, creating...')
            provisioning_config = DataFactoryCompute.provisioning_configuration()
            data_factory = ComputeTarget.create(workspace, factory_name, provisioning_config)
            data_factory.wait_for_completion()
            return data_factory
        else:
            raise e

data_factory_compute = get_or_create_data_factory(ws, data_factory_name)

# Assuming output data is your output from the step that you want to copy

transfer_to_known_location = DataTransferStep(
    name="transfer_to_known_location",
    source_data_reference=[output_data],
    destination_data_reference=blob_data_ref,
    compute_target=data_factory_compute
    )

from azureml.pipeline.core import Pipeline
from azureml.core import Workspace, Experiment

pipeline_01 = Pipeline(
    description="transfer_to_known_location",
    workspace=ws,
    steps=[transfer_to_known_location])

pipeline_run_01 = Experiment(ws, "transfer_to_known_location").submit(pipeline_01)
pipeline_run_01.wait_for_completion()
user_109522330
Related