Как сохранить встроенный преобразователь в большой двоичный объект, чтобы конвейер прогнозирования мог использовать его в службе AML?

Это, вероятно, недостаточно гибко для ваших нужд (к тому же, я еще не тестировал это), но если вы используете scikit-learn, одна из возможностей — включить шаг tf-idf/transformation в объект scikit-learn Pipeline и зарегистрировать это в ваше рабочее пространство.

Таким образом, ваш сценарий обучения будет содержать:

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')

и ваш сценарий отправки эксперимента будет содержать

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

Затем вы можете использовать зарегистрированную «модель» и развернуть ее как службу как поясняется в документации, загрузив ее в сценарий оценки через

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

Однако это приведет к преобразованию в вашем конвейере и, следовательно, не будет таким модульным, как вы просите...

Другое решение — передать DataReference в качестве входных данных для вашего PythonScriptStep.

Затем внутри transform.py вы можете прочитать это DataReference как аргумент командной строки.

Вы можете разобрать его и использовать как любой обычный путь для сохранения вашего векторизатора.

Например. ты сможешь:

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)

Затем внутри вашего скрипта (transform.py в приведенном выше примере) вы можете, например:

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')

ДОПОЛНИТЕЛЬНО: третий способ — просто зарегистрировать векторизатор как другую модель в вашем рабочем пространстве. Затем вы можете использовать его точно так же, как любую другую зарегистрированную модель. (Хотя этот параметр не требует явной записи в большой двоичный объект, как указано в вопросе выше)

Другой вариант — использовать DataTransferStep и использовать его для копирования вывода в «известное место». Этот блокнот содержит примеры использования DataTransferStep для копирования данных из различных поддерживаемых хранилищ данных и в них.

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()