<jupyter_start><jupyter_text>Licensed under the MIT License. ![Impressions](https://PixelServer20190423114238.azurewebsites.net/api/impressions/MachineLearningNotebooks/how-to-use-azureml/machine-learning-pipelines/parallel-run/tabular-dataset-partition-per-column.png)  Using Azure Machine Learning Pipelines and Parallel Run Step (PRS) for Many Model Training and Batch Inference using tabular input partitioned by column value


This example will create a partitioned tabular dataset by splitting the rows in a large csv file by its value on specified column. Each partition will form up a mini-batch in the parallel processing procedure.

The outline of this notebook is as follows:

- Create a tabular dataset partitioned by value on specified column.
- Do ML training of forecast model per each partition
- Do batch inference on the dataset with each mini-batch corresponds to one partition.
  Connect to workspace<jupyter_code>from azureml.core.workspace import Workspace

from azureml.core.authentication import InteractiveLoginAuthentication
from azureml.core import Workspace

subscription_id = 'YOUR_SUBSCRIPTION_ID'
# Azure Machine Learning resource group NOT the managed resource group
resource_group = 'YOUR_RESOURCE_GROUP' 

#Azure Machine Learning workspace name, NOT Azure Databricks workspace
workspace_name = 'YOUR_WORKSPACE'  
#Add tenant ID in case you have mutiple tenants
tenant_id ='YOUR_TENANT_ID' 

auth = InteractiveLoginAuthentication(tenant_id =tenant_id)
# Instantiate Azure Machine Learning workspace
ws = Workspace.get(name=workspace_name,
                   subscription_id=subscription_id,
                   resource_group=resource_group,auth= auth)

print('Workspace name: ' + ws.name, 
      'Azure region: ' + ws.location, 
      'Subscription id: ' + ws.subscription_id, 
      'Resource group: ' + ws.resource_group, sep = '\n')

datastore = ws.get_default_datastore()
from azureml.core.authentication import MsiAuthentication,ServicePrincipalAuthentication,TokenAuthentication, Audience
msi_auth = MsiAuthentication()
msi_auth.

import azureml.core
print(azureml.core.VERSION)<jupyter_output>1.31.0<jupyter_text>Download OJ sales data from opendataset url<jupyter_code># ! pip install azureml-opendatasets --user
import os
from azureml.opendatasets import OjSalesSimulated
dataset_maxfiles = 10 # Set to 11973 or 0 to get all the files

# Pull all of the data
oj_sales_files = OjSalesSimulated.get_file_dataset()

# Pull only the first `dataset_maxfiles` files
if dataset_maxfiles:
    oj_sales_files = oj_sales_files.take(dataset_maxfiles)

# Create a folder to download
download_path = '..\..\data\oj_sales_data' 
os.makedirs(download_path, exist_ok=True)

# Download the data
oj_sales_files.download(download_path, overwrite=True,)<jupyter_output><empty_output><jupyter_text>Upload OJ sales data to datastore<jupyter_code>target_path = 'oj_sales_data'

datastore.upload(src_dir = download_path,
                target_path = target_path,
                overwrite = True, 
                show_progress = True)<jupyter_output><empty_output><jupyter_text>Create tabular datasetCreate normal tabular dataset<jupyter_code>from azureml.core import Dataset

dataset = Dataset.Tabular.from_delimited_files(path=(datastore, 'oj_sales_data/*/*/*/*/*.csv'))
print(dataset.to_pandas_dataframe())<jupyter_output>WeekStarting  Store      Brand  Quantity  Advert  Price   Revenue
0      1990-06-14   1000  dominicks     12003       1   2.59  31087.77
1      1990-06-21   1000  dominicks     10239       1   2.39  24471.21
2      1990-06-28   1000  dominicks     17917       1   2.48  44434.16
3      1990-07-05   1000  dominicks     14218       1   2.33  33127.94
4      1990-07-12   1000  dominicks     15925       1   2.01  32009.25
...           ...    ...        ...       ...     ...    ...       ...
1205   1992-09-03   1003  dominicks     10302       1   1.94  19985.88
1206   1992-09-10   1003  dominicks     13502       1   2.16  29164.32
1207   1992-09-17   1003  dominicks     19644       1   2.67  52449.48
1208   1992-09-24   1003  dominicks     13860       1   2.29  31739.40
1209   1992-10-01   1003  dominicks     11040       1   1.99  21969.60

[1210 rows x 7 columns]<jupyter_text>Partition the tabular datasetPartition the dataset by column 'store' and 'brand'. You can get a partition of data by specifying the value of one or more partition keys. E.g., by specifying `store=1000 and brand='tropicana'`, you can get all the rows that matches this condition in the dataset.<jupyter_code>partitioned_dataset = dataset.partition_by(partition_keys=['Store', 'Brand'], target=(datastore, "partition_by_key_res"), name="partitioned_oj_data")
partitioned_dataset.partition_keys<jupyter_output>WARNING - Method partition_by: This is an experimental method, and may change at any time. Please see https://aka.ms/azuremlexperimental for more information.<jupyter_text>Create or Attach existing compute resource<jupyter_code>import os
from azureml.core.compute import AmlCompute, ComputeTarget

# choose a name for your cluster
compute_name = os.environ.get("AML_COMPUTE_CLUSTER_NAME", "cpu-cluster")
compute_min_nodes = os.environ.get("AML_COMPUTE_CLUSTER_MIN_NODES", 0)
compute_max_nodes = os.environ.get("AML_COMPUTE_CLUSTER_MAX_NODES", 2)

# This example uses CPU VM. For using GPU VM, set SKU to STANDARD_NC6
vm_size = os.environ.get("AML_COMPUTE_CLUSTER_SKU", "STANDARD_D2_V2")


if compute_name in ws.compute_targets:
    compute_target = ws.compute_targets[compute_name]
    if compute_target and type(compute_target) is AmlCompute:
        print('found compute target. just use it. ' + compute_name)
else:
    print('creating a new compute target...')
    provisioning_config = AmlCompute.provisioning_configuration(vm_size = vm_size,
                                                                min_nodes = compute_min_nodes, 
                                                                max_nodes = compute_max_nodes)

    # create the cluster
    compute_target = ComputeTarget.create(ws, compute_name, provisioning_config)
    
    # can poll for a minimum number of nodes and for a specific timeout. 
    # if no min node count is provided it will use the scale settings for the cluster
    compute_target.wait_for_completion(show_output=True, min_node_count=None, timeout_in_minutes=20)
    
     # For a more detailed view of current AmlCompute status, use get_status()
    print(compute_target.get_status().serialize())<jupyter_output><empty_output><jupyter_text>Intermediate/Output Data<jupyter_code>from azureml.pipeline.core import Pipeline, PipelineData

output_dir = PipelineData(name="inferences", datastore=datastore)
scripts_folder = "..\..\code"
inference_script_file = "aml_prs\prediction.py"
train_script_file = "aml_prs\model_train.py"<jupyter_output><empty_output><jupyter_text>Build and run the model training  pipeline
 Specify the environment to run the script
You would need to specify the required private azureml packages in dependencies.<jupyter_code>from azureml.core import Environment
from azureml.core.runconfig import CondaDependencies, DEFAULT_CPU_IMAGE

batch_conda_deps = CondaDependencies.create(pip_packages=['sklearn', 'pandas', 'joblib', 'azureml-defaults', 'azureml-core', 'azureml-dataprep[fuse]'])
batch_env = Environment(name="many_models_environment")
batch_env.python.conda_dependencies = batch_conda_deps
batch_env.docker.base_image = DEFAULT_CPU_IMAGE<jupyter_output><empty_output><jupyter_text>Training 
  Create the configuration to wrap the training script
The parameter `partition_keys` is a list containing a subset of the dataset partition keys, specifying how is the input dataset partitioned. Each and every possible combination of values of partition_keys will form up a mini-batch. E.g., by specifying `partition_keys=['store', 'brand']` will result in mini-batches like `store=1000 && brand=tropicana`, `store=1000 && brand=dominicks`, `store=1001 && brand=dominicks`, ...<jupyter_code>from azureml.pipeline.steps import ParallelRunStep, ParallelRunConfig

# In a real-world scenario, you'll want to shape your process per node and nodes to fit your problem domain.
parallel_run_train_config = ParallelRunConfig(
    source_directory=scripts_folder,
    entry_script=train_script_file,  # the user script to run against each input
    partition_keys=['Store', 'Brand'],
    error_threshold=-1,
    output_action='append_row',
    append_row_file_name="training_output.txt",
    environment=batch_env,
    compute_target=compute_target, 
    node_count=2,
    run_invocation_timeout=600
)
parallel_run_train_step = ParallelRunStep(
    name='train',
    inputs=[partitioned_dataset.as_named_input("partitioned_tabular_input")],
    output=output_dir,
    parallel_run_config=parallel_run_train_config,
    allow_reuse=False
)
from azureml.core import Experiment
from azureml.pipeline.core import Pipeline

pipeline = Pipeline(workspace=ws, steps=[parallel_run_train_step])

pipeline_run = Experiment(ws, 'Many_Model_Forecast').submit(pipeline)
pipeline_run.wait_for_completion(show_output=True)<jupyter_output><empty_output><jupyter_text>View the training results
In the model_train.py file you can see that the ResultList with training metrics gets returned. These are written to the DataStore specified in the PipelineData object as the output data, which in this case is called inferences. This containers the outputs from all of the worker nodes used in the compute cluster. You can download this data to view the results ... below just filters to the first 10 rows<jupyter_code>import pandas as pd
import tempfile

batch_run = pipeline_run.find_step_run(parallel_run_train_step.name)[0]
batch_output = batch_run.get_output_data(output_dir.name)

target_dir = tempfile.mkdtemp()
batch_output.download(local_path=target_dir)
result_file = os.path.join(target_dir, batch_output.path_on_datastore, parallel_run_train_config.append_row_file_name)

df = pd.read_csv(result_file, delimiter=" ", header=None)

df.columns = [ "Store", "Brand", "mse", "mape", "rmse", "model_name"]
print("Train result has ", df.shape[0], " rows")
df.head(10)<jupyter_output>Train result has  10  rows<jupyter_text>Inference  Create the configuration to wrap the inference scriptThe parameter `partition_keys` is a list containing a subset of the dataset partition keys, specifying how is the input dataset partitioned. Each and every possible combination of values of partition_keys will form up a mini-batch. E.g., by specifying `partition_keys=['store', 'brand']` will result in mini-batches like `store=1000 && brand=tropicana`, `store=1000 && brand=dominicks`, `store=1001 && brand=dominicks`, ...<jupyter_code>from azureml.pipeline.steps import ParallelRunStep, ParallelRunConfig

# In a real-world scenario, you'll want to shape your process per node and nodes to fit your problem domain.
parallel_run_inference_config = ParallelRunConfig(
    source_directory=scripts_folder,
    entry_script=inference_script_file,  # the user script to run against each input
    partition_keys=['Store', 'Brand'],
    error_threshold= -1,
    output_action='append_row',
    append_row_file_name="prediction_output.txt",
    environment=batch_env,
    compute_target=compute_target, 
    node_count=2,
    run_invocation_timeout=600
)<jupyter_output>WARNING - Parameter partition_keys: This is an experimental parameter, and may change at any time. Please see https://aka.ms/azuremlexperimental for more information.<jupyter_text>Create the pipeline step<jupyter_code>parallel_run_inference_step = ParallelRunStep(
    name='forecast',
    inputs=[partitioned_dataset.as_named_input("partitioned_tabular_input")],
    output=output_dir,
    parallel_run_config=parallel_run_inference_config,
    allow_reuse=False
)<jupyter_output><empty_output><jupyter_text>Run the pipeline<jupyter_code>from azureml.core import Experiment
from azureml.pipeline.core import Pipeline

pipeline = Pipeline(workspace=ws, steps=[parallel_run_inference_step])

pipeline_run = Experiment(ws, 'Many_Model_Forecast').submit(pipeline)
pipeline_run.wait_for_completion(show_output=True)<jupyter_output><empty_output><jupyter_text>View the prediction results
In the prediction.py file you can see that the ResultList with prediction result gets returned. These are written to the DataStore specified in the PipelineData object as the output data, which in this case is called inferences. This containers the outputs from all of the worker nodes used in the compute cluster. You can download this data to view the results ... below just filters to the first 10 rows<jupyter_code>import pandas as pd
import tempfile

batch_run = pipeline_run.find_step_run(parallel_run_inference_step.name)[0]
batch_output = batch_run.get_output_data(output_dir.name)

target_dir = tempfile.mkdtemp()
batch_output.download(local_path=target_dir)
result_file = os.path.join(target_dir, batch_output.path_on_datastore, parallel_run_config.append_row_file_name)

df = pd.read_csv(result_file, delimiter=" ", header=None)
df.columns = ["WeekStarting", "Prediction", "Store", "Brand"]
print("Prediction has ", df.shape[0], " rows")
df.head(10)<jupyter_output>Prediction has  1210  rows
{'fraction_non_alphanumeric': 0.07842258570468295, 'fraction_numerical': 0.02980058256777952, 'mean_word_length': 4.5148270181219115, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 51, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '308486', 'license': 'MIT', 'n_tokens_mistral': 4360, 'n_tokens_neox': 4038, 'n_words': 1117, 'path': 'sa-dsml-many-models/code/aml_prs/prs_many_models.ipynb', 'repo_name': 'microsoft/csa-misc-utils', 'size': '13389'}