Konfigurowanie rozwiązania AutoML do trenowania modelu prognozowania szeregów czasowych przy użyciu zestawu SDK i interfejsu wiersza polecenia

Obiekt można utworzyć MLTable przy użyciu elementu mltable zestawu SDK języka Python:

import mltable

paths = [
    {'file': './train_data/timeseries_train.csv'}
]

train_table = mltable.from_delimited_files(paths)
train_table.save('./train_data')

Ten kod tworzy nowy plik . /train_data/MLTable, który zawiera format pliku i instrukcje ładowania.

Aby rozpocząć zadanie trenowania, zdefiniuj obiekt danych wejściowych przy użyciu zestawu SDK języka Python:

from azure.ai.ml.constants import AssetTypes
from azure.ai.ml import Input

# Training MLTable defined locally, with local data to be uploaded.
my_training_data_input = Input(
    type=AssetTypes.MLTABLE, path="./train_data"
)

Nowy obiekt można zdefiniować MLTable , kopiując następujący fragment kodu YAML do nowego pliku ./train_data/MLTable.

$schema: https://azuremlschemas.azureedge.net/latest/MLTable.schema.json

type: mltable
paths:
    - file: ./timeseries_train.csv

transformations:
    - read_delimited:
        delimiter: ','
        encoding: ascii

Rozpocznij tworzenie konfiguracji YAML dla zadania automatycznego uczenia maszynowego przy użyciu danych treningowych określonych w poniższym przykładzie:

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

# Training data MLTable for the AutoML job.
training_data:
    path: "./train_data"
    type: mltable

validation_data:
    # Optional validation data.

compute: # Compute for training job.
primary_metric: # Primary metric.  

target_column_name: # Target column name.
n_cross_validations: # Cross-validation setting.

limits:
    # Limit settings.

forecasting:
    # Forecasting-specific settings.

training:
    # Training settings. 

Więcej szczegółów do tej konfiguracji dodasz w kolejnych sekcjach tego artykułu. W tym przykładzie lokalizacja to ./automl-forecasting-job.yml.

Poniższy przykład tworzy klaster obliczeniowy o nazwie cpu-cluster.

from azure.ai.ml.entities import AmlCompute

# specify aml compute name.
cpu_compute_target = "cpu-cluster"

try:
    ml_client.compute.get(cpu_compute_target)
except Exception:
    print("Creating a new cpu compute target...")
    compute = AmlCompute(
        name=cpu_compute_target, size="STANDARD_D2_V2", min_instances=0, max_instances=4
    )
    ml_client.compute.begin_create_or_update(compute).result()

Nowe środowisko obliczeniowe o nazwie cpu-compute można utworzyć przy użyciu następującego polecenia interfejsu wiersza polecenia platformy Azure:

az ml compute create -n cpu-compute --type amlcompute --min-instances 0 --max-instances 4

Odwołuj się do obliczeń w definicji zadania w następujący sposób:

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

# Set training data MLTable for the AutoML job.
training_data:
    path: "./train_data"
    type: mltable

# Set compute for the training job to use. 
compute: azureml:cpu-compute

primary_metric: # Primary metric.  

target_column_name: # Target column name.
n_cross_validations: # Cross-validation setting.

limits:
    # Limit settings.

forecasting:
    # Forecasting-specific settings.

training:
    # Training settings.

Funkcje fabryki automatycznego uczenia maszynowego służą do konfigurowania zadań prognozowania w zestawie SDK języka Python. W poniższym przykładzie pokazano, jak utworzyć zadanie prognozowania, ustawiając metrykę podstawową i ustawiając limity w przebiegu trenowania:

from azure.ai.ml import automl

# Set forecasting variables.
# As needed, modify the variable values to run the snippet successfully.
forecasting_job = automl.forecasting(
    compute="cpu-compute",
    experiment_name="sdk-v2-automl-forecasting-job",
    training_data=my_training_data_input,
    target_column_name=target_column_name,
    primary_metric="normalized_root_mean_squared_error",
    n_cross_validations="auto",
)

# Set optional limits.
forecasting_job.set_limits(
    timeout_minutes=120,
    trial_timeout_minutes=30,
    max_concurrent_trials=4,
)

Skonfiguruj ogólne właściwości zadania AutoML, w tym:

• Podstawowa metryka.
• Nazwa kolumny docelowej w danych treningowych.
• Ustawienia krzyżowego sprawdzania poprawności.
• Limity zasobów w zadaniu.

Aby uzyskać więcej informacji, zobacz schemat yaML zadania polecenia prognozowania, parametry trenowania i limity.

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

training_data:
    path: "./train_data"
    type: mltable

compute: azureml:cpu-compute

# Settings for primary metric, target/label column name, cross validation.
primary_metric: normalized_root_mean_squared_error
target_column_name: <target_column_name>
n_cross_validations: auto

# Settings for training job limits on time, concurrency, and others.
limits:
    timeout_minutes: 120
    trial_timeout_minutes: 30
    max_concurrent_trials: 4

forecasting:
    # Forecasting-specific settings.

training:
    # Training settings.

Użyj metod ForecastingJob, aby skonfigurować następujące ustawienia:

# Forecasting-specific configuration.
forecasting_job.set_forecast_settings(
    time_column_name=time_column_name,
    forecast_horizon=24
)

Te ustawienia są konfigurowane w forecasting sekcji konfiguracji YAML zadania:

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

training_data:
    path: "./train_data"
    type: mltable

compute: azureml:cpu-compute

primary_metric: normalized_root_mean_squared_error
target_column_name: <target_column_name>
n_cross_validations: auto

limits:
    timeout_minutes: 120
    trial_timeout_minutes: 30
    max_concurrent_trials: 4

# Forecasting-specific settings.
# Set the horizon to 24 for this example. The horizon generally depends on the business scenario.
forecasting:
    time_column_name: <time_column_name>
    forecast_horizon: 24

training:
    # Training settings.

# Forecasting-specific configuration.
# Add time series IDs for store and brand.
forecasting_job.set_forecast_settings(
    ...,  # Other settings.
    time_series_id_column_names=['store', 'brand']
)

# Forecasting-specific settings.
# Add time series IDs for store and brand.
forecasting:
    # Other settings.
    time_series_id_column_names: ["store", "brand"]

# Only search ExponentialSmoothing and ElasticNet models.
forecasting_job.set_training(
    allowed_training_algorithms=["ExponentialSmoothing", "ElasticNet"]
)

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

training_data:
    path: "./train_data"
    type: mltable

compute: azureml:cpu-compute

primary_metric: normalized_root_mean_squared_error
target_column_name: <target_column_name>
n_cross_validations: auto

limits:
    timeout_minutes: 120
    trial_timeout_minutes: 30
    max_concurrent_trials: 4

forecasting:
    time_column_name: <time_column_name>
    forecast_horizon: 24

# Training settings.
# Only search ExponentialSmoothing and ElasticNet models.
training:
    allowed_training_algorithms: ["ExponentialSmoothing", "ElasticNet"]
    # Other training settings.

# Search over all model classes except Prophet.
forecasting_job.set_training(
    blocked_training_algorithms=["Prophet"]
)

# Training settings.
# Search over all model classes except Prophet.
training:
    blocked_training_algorithms: ["Prophet"]
    # Other training settings.

# Include TCNForecaster models in the model search.
forecasting_job.set_training(
    enable_dnn_training=True
)

# Training settings.
# Include TCNForecaster models in the model search.
training:
    enable_dnn_training: true
    # Other training settings.

forecasting_job.set_forecast_settings(
    ...,  # Other settings.
    target_lags='auto', 
    target_rolling_window_size='auto',
    feature_lags='auto'
)

# Forecasting-specific settings.
# Auto configure lags and rolling-window features.
forecasting:
    target_lags: auto
    target_rolling_window_size: auto
    feature_lags: auto
    # Other settings.

forecasting_job.set_forecast_settings(
    ...,  # Other settings.
    short_series_handling_config='pad'
)

# Forecasting-specific settings.
# Auto configure lags and rolling-window features.
forecasting:
    short_series_handling_config: pad
    # Other settings.

# Aggregate the data to hourly frequency.
forecasting_job.set_forecast_settings(
    ...,  # Other settings.
    frequency='H',
    target_aggregate_function='sum'
)

# Forecasting-specific settings.
# Auto-configure lags and rolling-window features.
forecasting:
    frequency: H
    target_aggregate_function: sum
    # Other settings.

from azure.ai.ml import automl

# Create a job with five CV folds.
forecasting_job = automl.forecasting(
    ...,  # Other training parameters.
    n_cross_validations=5,
)

# Set the step size between folds to seven days.
forecasting_job.set_forecast_settings(
    ...,  # Other settings.
    cv_step_size=7
)

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

training_data:
    path: "./train_data"
    type: mltable

compute: azureml:cpu-compute

primary_metric: normalized_root_mean_squared_error
target_column_name: <target_column_name>
n_cross_validations: auto

# Use five CV folds.
n_cross_validations: 5

# Set the step size between folds to seven days.
forecasting:
    cv_step_size: 7
    # Other settings.

limits:
    # Limit settings.

training:
    # Training settings.

from azure.ai.ml.automl import ColumnTransformer

# Customize imputation methods for price and is_on_sale features.
# Median value imputation for price, constant value of zero for is_on_sale.
transformer_params = {
    "imputer": [
        ColumnTransformer(fields=["price"], parameters={"strategy": "median"}),
        ColumnTransformer(fields=["is_on_sale"], parameters={"strategy": "constant", "fill_value": 0}),
    ],
}

# Set the featurization.
# Ensure product_type feature is interpreted as categorical.
forecasting_job.set_featurization(
    mode="custom",
    transformer_params=transformer_params,
    column_name_and_types={"product_type": "Categorical"},
)

$schema: https://azuremlsdk2.blob.core.windows.net/preview/0.0.1/autoMLJob.schema.json
type: automl

experiment_name: cli-v2-automl-forecasting-job
description: A time-series forecasting AutoML job
task: forecasting

training_data:
    path: "./train_data"
    type: mltable

compute: azureml:cpu-compute

primary_metric: normalized_root_mean_squared_error
target_column_name: <target_column_name>
n_cross_validations: auto

# Customize imputation methods for price and is_on_sale features.
# Median value imputation for price, constant value of zero for is_on_sale.
featurization:
    mode: custom
    column_name_and_types:
        product_type: Categorical
    transformer_params:
        imputer:
            - fields: ["price"]
            parameters:
                strategy: median
            - fields: ["is_on_sale"]
            parameters:
                strategy: constant
                fill_value: 0

forecasting:
    # Forecasting-specific settings.

limits:
    # Limit settings.

training:
    # Training settings.

# Submit the AutoML job.
returned_job = ml_client.jobs.create_or_update(
    forecasting_job
)

print(f"Created job: {returned_job}")

# Get a URL for the job in the studio UI.
returned_job.services["Studio"].endpoint

W poniższym poleceniu Azure CLI, konfiguracja YAML zadania znajduje się w bieżącym katalogu roboczym pod ścieżką ./automl-forecasting-job.yml. Jeśli uruchomisz polecenie z innego katalogu, musisz odpowiednio zmienić ścieżkę.

run_id=$(az ml job create --file automl-forecasting-job.yml)

Możesz użyć przechowywanego identyfikatora przebiegu, aby zwrócić informacje o zadaniu. Parametr --web otwiera internetowy interfejs użytkownika usługi Azure Machine Learning Studio, w którym można wyświetlić szczegółowe informacje o zadaniu:

az ml job show -n $run_id --web

Poniższy przykład pobiera kod składnika z rejestru klienta:

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

# Get credential to access azureml registry.
try:
    credential = DefaultAzureCredential()
    # Check if token can be obtained successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential fails.
    credential = InteractiveBrowserCredential()

# Create client to access assets in azureml-preview registry.
ml_client_registry = MLClient(
    credential=credential,
    registry_name="azureml-preview"
)

# Create client to access assets in azureml registry.
ml_client_metrics_registry = MLClient(
    credential=credential,
    registry_name="azureml"
)

# Get inference component from registry.
inference_component = ml_client_registry.components.get(
    name="automl_forecasting_inference",
    label="latest"
)

# Get component to compute evaluation metrics from registry.
compute_metrics_component = ml_client_metrics_registry.components.get(
    name="compute_metrics",
    label="latest"
)

Następnie zdefiniuj funkcję fabryki, która tworzy potoki organizujące trenowanie, wnioskowanie i obliczanie metryk. Aby uzyskać więcej informacji, zobacz Konfigurowanie eksperymentu.

from azure.ai.ml import automl
from azure.ai.ml.constants import AssetTypes
from azure.ai.ml.dsl import pipeline

@pipeline(description="AutoML Forecasting Pipeline")
def forecasting_train_and_evaluate_factory(
    train_data_input,
    test_data_input,
    target_column_name,
    time_column_name,
    forecast_horizon,
    primary_metric='normalized_root_mean_squared_error',
    cv_folds='auto'
):
    # Configure training node of pipeline.
    training_node = automl.forecasting(
        training_data=train_data_input,
        target_column_name=target_column_name,
        primary_metric=primary_metric,
        n_cross_validations=cv_folds,
        outputs={"best_model": Output(type=AssetTypes.MLFLOW_MODEL)},
    )

    training_node.set_forecasting_settings(
        time_column_name=time_column_name,
        forecast_horizon=max_horizon,
        frequency=frequency,
        # Other settings.
        ... 
    )
    
    training_node.set_training(
        # Training parameters.
        ...
    )
    
    training_node.set_limits(
        # Limit settings.
        ...
    )

    # Configure inference node to make rolling forecasts on test set.
    inference_node = inference_component(
        test_data=test_data_input,
        model_path=training_node.outputs.best_model,
        target_column_name=target_column_name,
        forecast_mode='rolling',
        step=1
    )

    # Configure metrics calculation node.
    compute_metrics_node = compute_metrics_component(
        task="tabular-forecasting",
        ground_truth=inference_node.outputs.inference_output_file,
        prediction=inference_node.outputs.inference_output_file,
        evaluation_config=inference_node.outputs.evaluation_config_output_file
    )

    # Return dictionary with evaluation metrics and raw test set forecasts.
    return {
        "metrics_result": compute_metrics_node.outputs.evaluation_result,
        "rolling_fcst_result": inference_node.outputs.inference_output_file
    }

Zdefiniuj dane wejściowe trenowania i testowania zawarte w folderach lokalnych ./train_data i ./test_data.

my_train_data_input = Input(
    type=AssetTypes.MLTABLE,
    path="./train_data"
)

my_test_data_input = Input(
    type=AssetTypes.URI_FOLDER,
    path='./test_data',
)

Na koniec skonstruuj potok, ustaw jego domyślne obliczenia i prześlij zadanie:

pipeline_job = forecasting_train_and_evaluate_factory(
    my_train_data_input,
    my_test_data_input,
    target_column_name,
    time_column_name,
    forecast_horizon
)

# Set pipeline-level compute.
pipeline_job.settings.default_compute = compute_name

# Submit pipeline job.
returned_pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job,
    experiment_name=experiment_name
)
returned_pipeline_job

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

description: AutoML Forecasting Pipeline
experiment_name: cli-v2-automl-forecasting-pipeline

# Set default compute for pipeline steps.
settings:
    default_compute: cpu-compute

# Pipeline inputs.
inputs:
    train_data_input:
        type: mltable
        path: "./train_data"
    test_data_input:
        type: uri_folder
        path: "./test_data"
    target_column_name: <target column name>
    time_column_name: <time column name>
    forecast_horizon: <forecast horizon>
    primary_metric: normalized_root_mean_squared_error
    cv_folds: auto

# Set pipeline outputs.
# Output the evaluation metrics and raw test set rolling forecasts.
outputs: 
    metrics_result:
        type: uri_file
        mode: upload
    rolling_fcst_result:
        type: uri_file
        mode: upload

jobs:
  # Configure automl training node of pipeline.
    training_node:
        type: automl
        task: forecasting
        primary_metric: ${{parent.inputs.primary_metric}}
        target_column_name: ${{parent.inputs.target_column_name}}
        training_data: ${{parent.inputs.train_data_input}}
        n_cross_validations: ${{parent.inputs.cv_folds}}
        training:
            # Training settings.
        forecasting:
            time_column_name: ${{parent.inputs.time_column_name}}
            forecast_horizon: ${{parent.inputs.forecast_horizon}}
            # Other forecasting-specific settings.
        limits:
            # Limit settings.
        outputs:
            best_model:
                type: mlflow_model

    # Configure inference node to make rolling forecasts on test set.
    inference_node:
        type: command
        component: azureml://registries/azureml-preview/components/automl_forecasting_inference
        inputs:
            target_column_name: ${{parent.inputs.target_column_name}}
            forecast_mode: rolling
            step: 1
            test_data: ${{parent.inputs.test_data_input}}
            model_path: ${{parent.jobs.training_node.outputs.best_model}}
        outputs:
            inference_output_file: ${{parent.outputs.rolling_fcst_result}}
            evaluation_config_output_file:
                type: uri_file

    # Configure metrics calculation node.
    compute_metrics:
        type: command
        component: azureml://registries/azureml/compute_metrics
        inputs:
            task: "tabular-forecasting"
            ground_truth: ${{parent.jobs.inference_node.outputs.inference_output_file}}
            prediction: ${{parent.jobs.inference_node.outputs.inference_output_file}}
            evaluation_config: ${{parent.jobs.inference_node.outputs.evaluation_config_output_file}}
        outputs:
            evaluation_result: ${{parent.outputs.metrics_result}}

Rozwiązanie AutoML wymaga danych szkoleniowych w formacie MLTable.

Uruchom potok, używając następującego polecenia. Konfiguracja potoku znajduje się w ścieżce ./automl-forecasting-pipeline.yml:

run_id=$(az ml job create --file automl-forecasting-pipeline.yml -w <Workspace> -g <Resource Group> --subscription <Subscription>)

# Download metrics JSON.
ml_client.jobs.download(returned_pipeline_job.name, download_path=".", output_name='metrics_result')

# Download rolling forecasts.
ml_client.jobs.download(returned_pipeline_job.name, download_path=".", output_name='rolling_fcst_result')

az ml job download --name $run_id --download-path . --output-name metrics_result
az ml job download --name $run_id --download-path . --output-name rolling_fcst_result

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

# Get credential to access azureml registry.
try:
    credential = DefaultAzureCredential()
    # Check whether token can be obtained.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential if DefaultAzureCredential fails.
    credential = InteractiveBrowserCredential()

# Get many-models training component.
mm_train_component = ml_client_registry.components.get(
    name='automl_many_models_training',
    version='latest'
)

# Get many-models inference component.
mm_inference_component = ml_client_registry.components.get(
    name='automl_many_models_inference',
    version='latest'
)

# Get component to compute evaluation metrics.
compute_metrics_component = ml_client_metrics_registry.components.get(
    name="compute_metrics",
    label="latest"
)

@pipeline(description="AutoML Many Models Forecasting Pipeline")
def many_models_train_evaluate_factory(
    train_data_input,
    test_data_input,
    automl_config_input,
    compute_name,
    max_concurrency_per_node=4,
    parallel_step_timeout_in_seconds=3700,
    max_nodes=4,
    retrain_failed_model=False,
    forecast_mode="rolling",
    forecast_step=1
):
    mm_train_node = mm_train_component(
        raw_data=train_data_input,
        automl_config=automl_config_input,
        max_nodes=max_nodes,
        max_concurrency_per_node=max_concurrency_per_node,
        parallel_step_timeout_in_seconds=parallel_step_timeout_in_seconds,
        retrain_failed_model=retrain_failed_model,
        compute_name=compute_name
    )

    mm_inference_node = mm_inference_component(
        raw_data=test_data_input,
        max_nodes=max_nodes,
        max_concurrency_per_node=max_concurrency_per_node,
        parallel_step_timeout_in_seconds=parallel_step_timeout_in_seconds,
        optional_train_metadata=mm_train_node.outputs.run_output,
        forecast_mode=forecast_mode,
        step=forecast_step,
        compute_name=compute_name
    )

    compute_metrics_node = compute_metrics_component(
        task="tabular-forecasting",
        prediction=mm_inference_node.outputs.evaluation_data,
        ground_truth=mm_inference_node.outputs.evaluation_data,
        evaluation_config=mm_inference_node.outputs.evaluation_configs
    )

    # Return metrics results from rolling evaluation.
    return {
        "metrics_result": compute_metrics_node.outputs.evaluation_result
    }

Skonstruuj potok za pomocą funkcji factory. Dane szkoleniowe i testowe znajdują się w folderach lokalnych ./data/train i ./data/test. Na koniec ustaw domyślne obliczenia i prześlij zadanie, jak pokazano w poniższym przykładzie:

pipeline_job = many_models_train_evaluate_factory(
    train_data_input=Input(
        type="uri_folder",
        path="./data/train"
    ),
    test_data_input=Input(
        type="uri_folder",
        path="./data/test"
    ),
    automl_config=Input(
        type="uri_file",
        path="./automl_settings_mm.yml"
    ),
    compute_name="<cluster name>"
)
pipeline_job.settings.default_compute = "<cluster name>"

returned_pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job,
    experiment_name=experiment_name,
)
ml_client.jobs.stream(returned_pipeline_job.name)

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

description: AutoML Many Models Forecasting Pipeline
experiment_name: cli-v2-automl-mm-forecasting-pipeline

# Set default compute for pipeline steps.
settings:
    default_compute: azureml:cpu-compute

# Set pipeline inputs.
inputs:
    train_data_input:
        type: uri_folder
        path: "./train_data"
        mode: direct
    test_data_input:
        type: uri_folder
        path: "./test_data"
    automl_config_input:
        type: uri_file
        path: "./automl_settings_mm.yml"
    max_nodes: 4
    max_concurrency_per_node: 4
    parallel_step_timeout_in_seconds: 3700
    forecast_mode: rolling
    step: 1
    retrain_failed_model: False

# Set pipeline outputs.
# Output the evaluation metrics and raw test set rolling forecasts.
outputs: 
    metrics_result:
        type: uri_file
        mode: upload

jobs:
    # Configure AutoML many-models training component.
    mm_train_node:
        type: command
        component: azureml://registries/azureml-preview/components/automl_many_models_training
        inputs:
            raw_data: ${{parent.inputs.train_data_input}}
            automl_config: ${{parent.inputs.automl_config_input}}
            max_nodes: ${{parent.inputs.max_nodes}}
            max_concurrency_per_node: ${{parent.inputs.max_concurrency_per_node}}
            parallel_step_timeout_in_seconds: ${{parent.inputs.parallel_step_timeout_in_seconds}}
            retrain_failed_model: ${{parent.inputs.retrain_failed_model}}
        outputs:
            run_output:
                type: uri_folder

    # Configure inference node to make rolling forecasts on test set.
    mm_inference_node:
        type: command
        component: azureml://registries/azureml-preview/components/automl_many_models_inference
        inputs:
            raw_data: ${{parent.inputs.test_data_input}}
            max_concurrency_per_node: ${{parent.inputs.max_concurrency_per_node}}
            parallel_step_timeout_in_seconds: ${{parent.inputs.parallel_step_timeout_in_seconds}}
            forecast_mode: ${{parent.inputs.forecast_mode}}
            step: ${{parent.inputs.step}}
            max_nodes: ${{parent.inputs.max_nodes}}
            optional_train_metadata: ${{parent.jobs.mm_train_node.outputs.run_output}}
        outputs:
            run_output:
                type: uri_folder
            evaluation_configs:
                type: uri_file
            evaluation_data:
                type: uri_file

    # Configure metrics calculation node.
    compute_metrics:
        type: command
        component: azureml://registries/azureml/components/compute_metrics
        inputs:
            task: "tabular-forecasting"
            ground_truth: ${{parent.jobs.mm_inference_node.outputs.evaluation_data}}
            prediction: ${{parent.jobs.mm_inference_node.outputs.evaluation_data}}
            evaluation_config: ${{parent.jobs.mm_inference_node.outputs.evaluation_configs}}
        outputs:
            evaluation_result: ${{parent.outputs.metrics_result}}

Zadanie potoku danych należy uruchomić za pomocą następującego polecenia. Konfiguracja potoku obejmującego wiele modeli znajduje się w ścieżce ./automl-mm-forecasting-pipeline.yml:

az ml job create --file automl-mm-forecasting-pipeline.yml -w <Workspace> -g <Resource Group> --subscription <Subscription>

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

# Get credential to access azureml registry.
try:
    credential = DefaultAzureCredential()
    # Check whether token can be obtained.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential if DefaultAzureCredential fails.
    credential = InteractiveBrowserCredential()

# Get HTS training component.
hts_train_component = ml_client_registry.components.get(
    name='automl_hts_training',
    version='latest'
)

# Get HTS inference component.
hts_inference_component = ml_client_registry.components.get(
    name='automl_hts_inference',
    version='latest'
)

# Get component to compute evaluation metrics.
compute_metrics_component = ml_client_metrics_registry.components.get(
    name="compute_metrics",
    label="latest"
)

@pipeline(description="AutoML HTS Forecasting Pipeline")
def hts_train_evaluate_factory(
    train_data_input,
    test_data_input,
    automl_config_input,
    max_concurrency_per_node=4,
    parallel_step_timeout_in_seconds=3700,
    max_nodes=4,
    forecast_mode="rolling",
    forecast_step=1,
    forecast_level="SKU",
    allocation_method='proportions_of_historical_average'
):
    hts_train = hts_train_component(
        raw_data=train_data_input,
        automl_config=automl_config_input,
        max_concurrency_per_node=max_concurrency_per_node,
        parallel_step_timeout_in_seconds=parallel_step_timeout_in_seconds,
        max_nodes=max_nodes
    )
    hts_inference = hts_inference_component(
        raw_data=test_data_input,
        max_nodes=max_nodes,
        max_concurrency_per_node=max_concurrency_per_node,
        parallel_step_timeout_in_seconds=parallel_step_timeout_in_seconds,
        optional_train_metadata=hts_train.outputs.run_output,
        forecast_level=forecast_level,
        allocation_method=allocation_method,
        forecast_mode=forecast_mode,
        step=forecast_step
    )
    compute_metrics_node = compute_metrics_component(
        task="tabular-forecasting",
        prediction=hts_inference.outputs.evaluation_data,
        ground_truth=hts_inference.outputs.evaluation_data,
        evaluation_config=hts_inference.outputs.evaluation_configs
    )

    # Return metrics results from rolling evaluation.
    return {
        "metrics_result": compute_metrics_node.outputs.evaluation_result
    }

Konstruowanie potoku przy użyciu funkcji factory. Dane szkoleniowe i testowe znajdują się w folderach lokalnych ./data/train i ./data/test. Na koniec ustaw domyślne obliczenia i prześlij zadanie, jak pokazano w poniższym przykładzie:

pipeline_job = hts_train_evaluate_factory(
    train_data_input=Input(
        type="uri_folder",
        path="./data/train"
    ),
    test_data_input=Input(
        type="uri_folder",
        path="./data/test"
    ),
    automl_config=Input(
        type="uri_file",
        path="./automl_settings_hts.yml"
    )
)
pipeline_job.settings.default_compute = "cluster-name"

returned_pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job,
    experiment_name=experiment_name,
)
ml_client.jobs.stream(returned_pipeline_job.name)

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

description: AutoML Many Models Forecasting Pipeline
experiment_name: cli-v2-automl-mm-forecasting-pipeline

# Set the default compute for pipeline steps.
settings:
    default_compute: cpu-compute

# Set pipeline inputs.
inputs:
    train_data_input:
        type: uri_folder
        path: "./train_data"
        mode: direct
    test_data_input:
        type: uri_folder
        path: "./test_data"
    automl_config_input:
        type: uri_file
        path: "./automl_settings_hts.yml"
    max_concurrency_per_node: 4
    parallel_step_timeout_in_seconds: 3700
    max_nodes: 4
    forecast_mode: rolling
    step: 1
    allocation_method: proportions_of_historical_average
    forecast_level: # forecast level

# Set pipeline outputs.
# Output evaluation metrics and raw test set rolling forecasts.
outputs: 
    metrics_result:
        type: uri_file
        mode: upload

jobs:
    # Configure AutoML many-models training component.
    hts_train_node:
        type: command
        component: azureml://registries/azureml-preview/components/automl_hts_training
        inputs:
            raw_data: ${{parent.inputs.train_data_input}}
            automl_config: ${{parent.inputs.automl_config_input}}
            max_nodes: ${{parent.inputs.max_nodes}}
            max_concurrency_per_node: ${{parent.inputs.max_concurrency_per_node}}
            parallel_step_timeout_in_seconds: ${{parent.inputs.parallel_step_timeout_in_seconds}}
        outputs:
            run_output:
                type: uri_folder


    # Configure inference node to make rolling forecasts on test set.
    hts_inference_node:
        type: command
        component: azureml://registries/azureml-preview/components/automl_hts_inference
        inputs:
            raw_data: ${{parent.inputs.test_data_input}}
            max_concurrency_per_node: ${{parent.inputs.max_concurrency_per_node}}
            parallel_step_timeout_in_seconds: ${{parent.inputs.parallel_step_timeout_in_seconds}}
            forecast_mode: ${{parent.inputs.forecast_mode}}
            step: ${{parent.inputs.step}}
            max_nodes: ${{parent.inputs.max_nodes}}
            optional_train_metadata: ${{parent.jobs.hts_train_node.outputs.run_output}}
            forecast_level: ${{parent.inputs.forecast_level}}
            allocation_method: ${{parent.inputs.allocation_method}}
        outputs:
            run_output:
                type: uri_folder
            evaluation_configs:
                type: uri_file
            evaluation_data:
                type: uri_file

    # Configure metrics calculation node.
    compute_metrics:
        type: command
        component: azureml://registries/azureml/components/compute_metrics
        inputs:
            task: "tabular-forecasting"
            ground_truth: ${{parent.jobs.hts_inference_node.outputs.evaluation_data}}
            prediction: ${{parent.jobs.hts_inference_node.outputs.evaluation_data}}
            evaluation_config: ${{parent.jobs.hts_inference_node.outputs.evaluation_configs}}
        outputs:
            evaluation_result: ${{parent.outputs.metrics_result}}

Uruchamiasz zadanie potoku przy użyciu następującego polecenia. Konfiguracja potoku wielomodelowego znajduje się w ścieżce ./automl-hts-forecasting-pipeline.yml.

az ml job create --file automl-hts-forecasting-pipeline.yml -w <Workspace> -g <Resource Group> --subscription <Subscription>