Wallaroo MLOps API Essentials Guide: Pipeline Management

How to use the Wallaroo API for Pipeline Management

Pipeline Management

Pipelines can be managed through the Wallaroo API. Pipelines are the vehicle used for deploying, serving, and monitoring ML models. For more information, see the Wallaroo Glossary.

Create Pipeline in a Workspace

Creates a new pipeline in the specified workspace.

  • Parameters
    • pipeline_id - (REQUIRED string): Name of the new pipeline.
    • workspace_id - (REQUIRED int): Numerical id of the workspace for the new pipeline.
    • definition - (REQUIRED string): Pipeline definitions, can be {} for none.

Example: Two pipelines are created in the workspace created in the step Create Workspace. One will be an empty pipeline without any models, the other will be created using the uploaded models in the Upload Model to Workspace step and no configuration details. The pipeline id, variant id, and variant version of each pipeline will be stored for later examples.

# Create pipeline in a workspace

apiRequest = "/pipelines/create"

exampleEmptyPipelineName=f"emptypipeline-{uuid.uuid4()}"

data = {
  "pipeline_id": exampleEmptyPipelineName,
  "workspace_id": exampleWorkspaceId,
  "definition": {}
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
exampleEmptyPipelineId = response['pipeline_pk_id']
exampleEmptyPipelineVariantId=response['pipeline_variant_pk_id']
emptyExamplePipelineVariantVersion=['pipeline_variant_version']
response

{'pipeline_pk_id': 3,
 'pipeline_variant_pk_id': 3,
 'pipeline_variant_version': '84730f78-7b89-4420-bdcb-3c5abac0dd10'}

# Create pipeline in a workspace with models

apiRequest = "/pipelines/create"

exampleModelPipelineName=f"pipelinewithmodel-{uuid.uuid4()}"
exampleModelDeployName = f"deploywithmodel-{uuid.uuid4()}"

data = {
  "pipeline_id": exampleModelPipelineName,
  "workspace_id": exampleWorkspaceId,
  "definition": {
      "id":exampleModelDeployName,
      "steps":
      [
          {
          "ModelInference":
          {
              "models": [
                    {
                        "name":exampleModelName,
                        "version":exampleModelVersion,
                        "sha":exampleModelSha
                    }
                ]
          }
          }
      ]
  }
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
exampleModelPipelineId = response['pipeline_pk_id']
exampleModelPipelineVariantId=response['pipeline_variant_pk_id']
emptyModelPipelineVariantVersion=['pipeline_variant_version']
response

{'pipeline_pk_id': 108,
 'pipeline_variant_pk_id': 108,
 'pipeline_variant_version': '246685fd-258a-49a6-b431-ae4847890eee'}

Deploy a Pipeline

Deploy a an existing pipeline. Note that for any pipeline that has model steps, they must be included either in model_configs, model_ids or models.

  • Parameters
    • deploy_id (REQUIRED string): The name for the pipeline deployment.
    • engine_config (OPTIONAL string): Additional configuration options for the pipeline.
    • pipeline_version_pk_id (REQUIRED int): Pipeline version id.
    • model_configs (OPTIONALArray int): Ids of model configs to apply.
    • model_ids (OPTIONALArray int): Ids of models to apply to the pipeline. If passed in, model_configs will be created automatically.
    • models (OPTIONAL Array models): If the model ids are not available as a pipeline step, the models’ data can be passed to it through this method. The options below are only required if models are provided as a parameter.
      • name (REQUIRED string): Name of the uploaded model that is in the same workspace as the pipeline.
      • version (REQUIRED string): Version of the model to use.
      • sha (REQUIRED string): SHA value of the model.
    • pipeline_id (REQUIRED int): Numerical value of the pipeline to deploy.
  • Returns
    • id (int): The deployment id.

Examples: Both the empty pipeline and pipeline with model created in the step Create Pipeline in a Workspace will be deployed and their deployment information saved for later examples.

# Deploy empty pipeline

apiRequest = "/pipelines/deploy"

exampleEmptyDeployId = f"emptydeploy-{uuid.uuid4()}"

data = {
    "deploy_id": exampleEmptyDeployId,
    "pipeline_version_pk_id": exampleEmptyPipelineVariantId,
    "pipeline_id": exampleEmptyPipelineId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
exampleEmptyDeploymentId=response['id']
response

{'id': 2}

# Deploy a pipeline with models

apiRequest = "/pipelines/deploy"
exampleModelDeployId=f"modeldeploy-{uuid.uuid4()}"

data = {
    "deploy_id": exampleModelDeployId,
    "pipeline_version_pk_id": exampleModelPipelineVariantId,
    "models": [
        {
            "name":exampleModelName,
            "version":exampleModelVersion,
            "sha":exampleModelSha
        }
    ],
    "pipeline_id": exampleModelPipelineId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
exampleModelDeploymentId=response['id']
response

{'id': 60}

Get Deployment Status

Returns the deployment status.

  • Parameters
    • name - (REQUIRED string): The deployment in the format {deployment_name}-{deploymnent-id}.

Example: The deployed empty and model pipelines status will be displayed.

# Get empty pipeline deployment

apiRequest = "/status/get_deployment"

data = {
  "name": f"{exampleEmptyDeployId}-{exampleEmptyDeploymentId}"
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
response

# Get model pipeline deployment

apiRequest = "/status/get_deployment"

data = {
  "name": f"{exampleModelDeployId}-{exampleModelDeploymentId}"
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
response

{'status': 'Running',
 'details': [],
 'engines': [{'ip': '10.4.1.151',
   'name': 'engine-577db84597-x7bm4',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'pipelinewithmodel-94676967-b018-4002-89ef-1d69defc6273',
      'status': 'Running'}]},
   'model_statuses': {'models': [{'name': 'apitestmodel-dfa7e9fd-df72-4b28-93f6-3d147c9f962f',
      'version': '87476d85-b8ee-4714-81ba-53041b26f50f',
      'sha': 'bc85ce596945f876256f41515c7501c399fd97ebcb9ab3dd41bf03f8937b4507',
      'status': 'Running'}]}}],
 'engine_lbs': [{'ip': '10.4.2.59',
   'name': 'engine-lb-7d6f4bfdd-6hxj5',
   'status': 'Running',
   'reason': None,
   'details': []}],
 'sidekicks': []}

Get External Inference URL

The API command /admin/get_pipeline_external_url retrieves the external inference URL for a specific pipeline in a workspace.

  • Parameters
    • workspace_id (REQUIRED integer): The workspace integer id.
    • pipeline_name (REQUIRED string): The name of the deployment.

In this example, a list of the workspaces will be retrieved. Based on the setup from the Internal Pipeline Deployment URL Tutorial, the workspace matching urlworkspace will have it’s workspace id stored and used for the /admin/get_pipeline_external_url request with the pipeline urlpipeline.

The External Inference URL will be stored as a variable for the next step.

## Retrieve the pipeline's External Inference URL
TOKEN=get_jwt_token(TOKENURL, CLIENT, SECRET, USERNAME, PASSWORD)

apiRequest = "/admin/get_pipeline_external_url"

data = {
    "workspace_id": exampleWorkspaceId,
    "pipeline_name": exampleModelDeployId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
print(response)
externalUrl = response['url']
externalUrl

{'url': 'https://doc-test.api.example.com/v1/api/pipelines/infer/modeldeploy-15'}

‘https://doc-test.api.example.com/v1/api/pipelines/infer/modeldeploy-15'

Perform Inference Through External URL

The inference can now be performed through the External Inference URL. This URL will accept the same inference data file that is used with the Wallaroo SDK, or with an Internal Inference URL as used in the Internal Pipeline Inference URL Tutorial.

For this example, the externalUrl retrieved through the Get External Inference URL is used to submit a single inference request.

If the Wallaroo instance has been configured to enable Arrow support, then the file cc_data_1k.df.json will be used. This is a DataFrame object. If Arrow support has not been enabled, then the inference request is used with the Wallaroo proprietary JSON data file cc_data_1k.json.

#TOKEN=get_jwt_token(TOKENURL, CLIENT, SECRET, USERNAME, PASSWORD)
# if Arrow has been enabled, set this to True.  Otherwise, leave as False.
arrowEnabled = True
## Inference through external URL

# retrieve the json data to submit
if arrowEnabled is True:
    dataFile = './data/cc_data_1k.df.json'
    data = json.load(open('./data/cc_data_1k.df.json','rb'))
    contentType="application/json; format=pandas-records"
else:
    dataFile = './data/cc_data_1k.json'
    data = json.load(open('./data/cc_data_1k.json','rb'))
    contentType="application/json"

# set the headers
headers= {
    'Authorization': 'Bearer ' + TOKEN,
    'Content-Type': contentType
}

# submit the request via POST
response = requests.post(externalUrl, json=data, headers=headers)

# Only the first 300 characters will be displayed for brevity
printResponse = json.dumps(response.json())
print(printResponse[0:300])

[{"time": 1677788050393, "in": {"tensor": [-1.0603297501, 2.3544967095, -3.5638788326, 5.1387348926, -1.2308457019, -0.7687824608, -3.5881228109, 1.8880837663, -3.2789674274, -3.9563254554, 4.0993439118, -5.6539176395, -0.8775733373, -9.131571192, -0.6093537873, -3.7480276773, -5.0309125017, -0.8748

Undeploy a Pipeline

Undeploys a deployed pipeline.

  • Parameters
    • pipeline_id - (REQUIRED int): The numerical id of the pipeline.
    • deployment_id - (REQUIRED int): The numerical id of the deployment.
  • Returns
    • Nothing if the call is successful.

Example: Both the empty pipeline and pipeline with models deployed in the step Deploy a Pipeline will be undeployed.

# Undeploy an empty pipeline

apiRequest = "/pipelines/undeploy"

data = {
    "pipeline_id": exampleEmptyPipelineId,
    "deployment_id":exampleEmptyDeploymentId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
response

# Undeploy pipeline with models

apiRequest = "/pipelines/undeploy"

data = {
    "pipeline_id": exampleModelPipelineId,
    "deployment_id":exampleModelDeploymentId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
response

Copy a Pipeline

Copies an existing pipeline into a new one in the same workspace. A new engine configuration can be set for the copied pipeline.

  • Parameters
    • name - (REQUIRED string): The name of the new pipeline.
    • workspace_id - (REQUIRED int): The numerical id of the workspace to copy the source pipeline from.
    • source_pipeline - (REQUIRED int): The numerical id of the pipeline to copy from.
    • deploy - (OPTIONAL string): Name of the deployment.
    • engine_config - (OPTIONAL string): Engine configuration options.
    • pipeline_version - (OPTIONAL string): Optional version of the copied pipeline to create.

Example: The pipeline with models created in the step Create Pipeline in a Workspace will be copied into a new one.

# Copy a pipeline

apiRequest = "/pipelines/copy"

exampleCopiedPipelineName=f"copiedmodelpipeline-{uuid.uuid4()}"

data = {
  "name": exampleCopiedPipelineName,
  "workspace_id": exampleWorkspaceId,
  "source_pipeline": exampleModelPipelineId
}

response = get_wallaroo_response(APIURL, apiRequest, TOKEN, data)
response

{'pipeline_pk_id': 5,
 'pipeline_variant_pk_id': 5,
 'pipeline_version': None,
 'deployment': None}