【Azure AutoML】時系列予測をPythonで行う

train.py
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient, Input, automl
from azure.ai.ml.constants import AssetTypes
from azure.ai.ml.entities import AmlCompute

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=

ml_client = MLClient(credential, subscription_id, resource_group, workspace_name)

#データ
my_training_data_input = Input(
    type=AssetTypes.MLTABLE, path="azureml://datastores/workspaceblobstore/paths/"
)

#コンピューティングの作成
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=1
    )
    ml_client.compute.begin_create_or_update(compute).result()

forecasting_job = automl.forecasting(
    compute=cpu_compute_target,
    experiment_name="",
    training_data=my_training_data_input,
    target_column_name="",
    primary_metric="NormalizedRootMeanSquaredError",
    n_cross_validations=5,
    enable_model_explainability=True,
)

forecasting_job.set_limits(
    timeout_minutes=600,
    trial_timeout_minutes=100,
    max_trials=4,
)

forecasting_job.set_forecast_settings(
    time_column_name=,
    forecast_horizon=24,
)

forecasting_job.set_training(

# allowed_training_algorithms=["ExponentialSmoothing", "ElasticNet"]

# blocked_training_algorithms=["Prophet"]

)

returned_job = ml_client.jobs.create_or_update(forecasting_job)
print(f"Created job: {returned_job.name}")

The refresh token has expired due to inactivity.

paths:
  - file: day.csv
transformations:
  - read_delimited:
        delimiter: ','
        encoding: 'ascii'

Created job: musing_vinegar_w4cxgmzvlr

job_status.py
from azure.ai.ml.entities import Job
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=

ml_client = MLClient(credential, subscription_id, resource_group, workspace_name)

returned_job = ml_client.jobs.get(name="")
print(returned_job.status)
print(returned_job.tags["automl_best_child_run_id"])

returned_job = ml_client.jobs.get(name="musing_vinegar_w4cxgmzvlr")

Completed
musing_vinegar_w4cxgmzvlr_0

register.py
from azure.ai.ml.entities import Model
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=
ml_client = MLClient(
    credential, subscription_id, resource_group, workspace
)

run_model = Model(

    path="azureml://subscriptions//resourceGroups//workspaces//datastores/workspaceartifactstore/paths/ExperimentRun/dcid./outputs/mlflow-model/",
    name=""
)

ml_client.models.create_or_update(run_model)

run_model = Model(

    path="azureml://subscriptions//resourceGroups//workspaces//datastores/workspaceartifactstore/paths/ExperimentRun/dcid.musing_vinegar_w4cxgmzvlr_0/outputs/mlflow-model/",
    name="musing_vinegar_w4cxgmzvlr_0"
)

endpoint.py
from azure.ai.ml import MLClient
from azure.ai.ml.entities import ManagedOnlineEndpoint
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=
ml_client = MLClient(
    credential, subscription_id, resource_group, workspace
)

endpoint_name = ""
endpoint = ManagedOnlineEndpoint(name=endpoint_name)

endpoint_result=ml_client.online_endpoints.begin_create_or_update(endpoint)

endpoint_status.py
from azure.ai.ml.entities import OnlineEndpoint
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=

ml_client = MLClient(credential, subscription_id, resource_group, workspace_name)

endpoint = ml_client.online_endpoints.get(name="")
api_key = ml_client.online_endpoints.get_keys(name="").primary_key

print(endpoint.provisioning_state)
print(api_key)

Succeeded
XXX(APIキー)

sas_outputs.py
import os
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient, BlobSasPermissions,  generate_blob_sas
import datetime
import requests

try:
    account_url = "https://.blob.core.windows.net"
    credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)
    blob_service_client = BlobServiceClient(account_url, credential=credential)

    container_name = "azureml"
    container_client = blob_service_client.get_container_client(container=container_name)

    outputs_path = "ExperimentRun/dcid./outputs/"
    blob_list = container_client.list_blobs(name_starts_with=outputs_path)

    outputs_folder = "outputs"
    if not os.path.exists(outputs_folder):
        os.makedirs(outputs_folder)

    for blob in blob_list:
        sas_token=generate_blob_sas(
        blob_service_client.account_name,
        account_key="",
        container_name=container_name,
        blob_name=blob.name,
        permission=BlobSasPermissions(read=True),
        expiry=datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    )

        sas_url=f"{account_url}/{container_name}/{blob.name}?{sas_token}"
        response = requests.get(sas_url)
        print(sas_url)
        if response.status_code==200:

            file_path=os.path.join(outputs_folder,(blob.name.split("/")[-1]))
            with open(file_path, mode="wb") as download_file:
                download_file.write(response.content)
        else:
            print("Failed to download the file")

except Exception as ex:
    print("Exception:")
    print(ex)

deploy.py
from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineDeployment,
    Model,
    Environment,
    CodeConfiguration,
)
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=
ml_client = MLClient(
    credential, subscription_id, resource_group, workspace
)

local_endpoint_name = ""

model = Model(

    path="azureml://subscriptions//resourceGroups//workspaces//datastores/workspaceartifactstore/paths/ExperimentRun/dcid./outputs/mlflow-model/model.pkl",
    name=""
)

env = Environment(
    conda_file="outputs/conda_env_v_1_0_0.yml",
    image="mcr.microsoft.com/azureml/minimal-ubuntu20.04-py38-cpu-inference:latest",
)
deployment = ManagedOnlineDeployment(
    name="",
    endpoint_name=local_endpoint_name,
    model=model,
    environment=env,
    code_configuration=CodeConfiguration(
        code="outputs/",
        scoring_script="scoring_file_v_2_0_0.py"
    ),
    instance_type="Standard_DS2_v2",
    instance_count=1,
)

ml_client.online_deployments.begin_create_or_update(deployment=deployment)

deploy_status.py
from azure.ai.ml.entities import OnlineEndpoint
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=

ml_client = MLClient(credential, subscription_id, resource_group, workspace_name)

deployment = ml_client.online_deployments.get(name="",endpoint_name="")
print(deployment.provisioning_state)

Succeeded

predict.py
import urllib.request
import json
import os
import ssl

def allowSelfSignedHttps(allowed):

# bypass the server certificate verification on client side

    if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
        ssl._create_default_https_context = ssl._create_unverified_context

allowSelfSignedHttps(True) # this line is needed if you use self-signed certificate in your scoring service.

# Request data goes here

# The example below assumes JSON formatting which may be updated

# depending on the format your endpoint expects.

# More information can be found here:

# https://docs.microsoft.com/azure/machine-learning/how-to-deploy-advanced-entry-script

data =  {
  "Inputs": {
    "data": [
      {
        "datetime": "2012-12-20T00:00:00.000Z",
        "season": 4,
        "holiday": 0,
        "workingday": 1,
        "weather": 1,
        "temp": 13.12,
        "atemp": 17.425,
        "humidity": 66,
        "windspeed": 0.0,
        "casual": 0,
        "registered": 0
      }
    ]
  },
  "GlobalParameters": {
    "quantiles": [
      0.025,
      0.975
    ]
  }
}

body = str.encode(json.dumps(data))

url = 'https://XXX'

# Replace this with the primary/secondary key or AMLToken for the endpoint

api_key = '<APIキー>'
if not api_key:
    raise Exception("A key should be provided to invoke the endpoint")

# The azureml-model-deployment header will force the request to go to a specific deployment.

# Remove this header to have the request observe the endpoint traffic rules

headers = {'Content-Type':'application/json', 'Authorization':('Bearer '+ api_key), 'azureml-model-deployment': 'deploy-name2' }

req = urllib.request.Request(url, body, headers)

try:
    response = urllib.request.urlopen(req)

    result = response.read()
    print(result)
except urllib.error.HTTPError as error:
    print("The request failed with status code: " + str(error.code))

# Print the headers - they include the requert ID and the timestamp, which are useful for debugging the failure

    print(error.info())
    print(error.read().decode("utf8", 'ignore'))

b'{"Results": {"forecast": [68.58547751506723], "prediction_interval": ["[-147.55469732943465, 284.72565235956904]"], "index": [{"datetime": 1355961600000}]}}'

delete.py
from azure.identity import DefaultAzureCredential
from azure.ai.ml import MLClient

credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)

subscription_id=
resource_group=
workspace_name=

ml_client = MLClient(credential, subscription_id, resource_group, workspace_name)

returned_endpoint=ml_client.online_endpoints.begin_delete(name="")
returned_cluster=ml_client.compute.begin_delete(name="")

conda_env_v_1_0_0.yml

# Conda environment specification. The dependencies defined in this file will

# be automatically provisioned for runs with userManagedDependencies=False.

# Details about the Conda environment file format:

# https://conda.io/docs/user-guide/tasks/manage-environments.html#create-env-file-manually

name: project_environment
dependencies:

# The python interpreter version.

# Currently Azure ML only supports 3.8 and later.

- python=3.8.16

- pip:
  - azureml-train-automl-runtime==1.49.0
  - inference-schema
  - azureml-interpret==1.49.0
  - azureml-defaults==1.49.0
- numpy==1.21.6
- pandas==1.1.5
- scikit-learn==0.22.1
- py-xgboost==1.3.3
- fbprophet==0.7.1
- holidays==0.10.3
- psutil==5.9.4
channels:
- anaconda
- conda-forge

scoring_file_v_2_0_0.py

# ---------------------------------------------------------

# Copyright (c) Microsoft Corporation. All rights reserved.

# ---------------------------------------------------------

import json
import logging
import os
import pickle
import numpy as np
import pandas as pd
import joblib

import azureml.automl.core
from azureml.automl.core.shared import logging_utilities, log_server
from azureml.telemetry import INSTRUMENTATION_KEY

from inference_schema.schema_decorators import input_schema, output_schema
from inference_schema.parameter_types.numpy_parameter_type import NumpyParameterType
from inference_schema.parameter_types.pandas_parameter_type import PandasParameterType
from inference_schema.parameter_types.standard_py_parameter_type import StandardPythonParameterType

data_sample = PandasParameterType(pd.DataFrame({"datetime": pd.Series(["2000-1-1"], dtype="datetime64[ns]"), "season": pd.Series([0], dtype="int64"), "holiday": pd.Series([0], dtype="int64"), "workingday": pd.Series([0], dtype="int64"), "weather": pd.Series([0], dtype="int64"), "temp": pd.Series([0.0], dtype="float64"), "atemp": pd.Series([0.0], dtype="float64"), "humidity": pd.Series([0], dtype="int64"), "windspeed": pd.Series([0.0], dtype="float64"), "casual": pd.Series([0], dtype="int64"), "registered": pd.Series([0], dtype="int64")}), enforce_shape=False)
input_sample = StandardPythonParameterType({'data': data_sample})
quantiles_sample = StandardPythonParameterType([0.025, 0.975])
sample_global_params = StandardPythonParameterType({"quantiles": quantiles_sample})

result_sample = StandardPythonParameterType({
    'forecast': NumpyParameterType(np.array([0])),
    'prediction_interval': NumpyParameterType(np.array([0])),
    'index': PandasParameterType(pd.DataFrame({}), enforce_shape=False)
})
output_sample = StandardPythonParameterType({'Results': result_sample})
try:
    log_server.enable_telemetry(INSTRUMENTATION_KEY)
    log_server.set_verbosity('INFO')
    logger = logging.getLogger('azureml.automl.core.scoring_script_forecasting_v2')
except Exception:
    pass

def init():
    global model

# This name is model.id of model that we want to deploy deserialize the model file back

# into a sklearn model

    model_path = os.path.join(os.getenv('AZUREML_MODEL_DIR'), 'model.pkl')
    path = os.path.normpath(model_path)
    path_split = path.split(os.sep)
    log_server.update_custom_dimensions({'model_name': path_split[-3], 'model_version': path_split[-2]})
    try:
        logger.info("Loading model from path.")
        model = joblib.load(model_path)
        logger.info("Loading successful.")
    except Exception as e:
        logging_utilities.log_traceback(e, logger)
        raise

@input_schema('GlobalParameters', sample_global_params, convert_to_provided_type=False)
@input_schema('Inputs', input_sample)
@output_schema(output_sample)
def run(Inputs, GlobalParameters={"quantiles":[0.025, 0.975]}):
    y_query = None
    data = Inputs['data']
    if 'y_query' in data.columns:
        y_query = data.pop('y_query').values
    
    quantiles = GlobalParameters.get("quantiles", [0.025, 0.975])
    quantiles = [min(quantiles), 0.5, max(quantiles)]
    PI = 'prediction_interval'
    model.quantiles = quantiles
    pred_quantiles = model.forecast_quantiles(data, y_query)

    pred_quantiles[PI] = pred_quantiles[[min(quantiles), max(quantiles)]].apply(lambda x: '[{}, {}]'.format(x[0], x[1]), axis=1)

    index_as_df = pred_quantiles.iloc[:,:-4].reset_index(drop=True)# get time column name and grain column name
    forecast_as_list = pred_quantiles[0.5].to_list()
    PI_as_list = pred_quantiles[PI].to_list()
    result = {
        "forecast": forecast_as_list,
        "prediction_interval": PI_as_list,
        "index": json.loads(index_as_df.to_json(orient='records'))
    }
    return {'Results': result}