import wallaroo
from wallaroo.object import EntityNotFoundError

import pyarrow as pa
import requests

from IPython.display import display

# used to display DataFrame information without truncating
from IPython.display import display
import pandas as pd
pd.set_option('display.max_colwidth', None)
pd.set_option('display.max_columns', None)

import datetime

import os

# Login through local Wallaroo instance

wl = wallaroo.Client()

workspace_name = 'logworkspace'
main_pipeline_name = 'logpipeline-test'
model_name_control = 'logcontrol'
model_file_name_control = './models/rf_model.onnx'

workspace = wl.get_workspace(name=workspace_name, create_if_not_exist=True)

wl.set_current_workspace(workspace)

{'name': 'logworkspace', 'id': 10, 'archived': False, 'created_by': 'ee2bee02-d1f1-4302-916b-08a9d2bf88b1', 'created_at': '2025-02-06T17:24:30.687557+00:00', 'models': [], 'pipelines': []}

housing_model_control = (wl.upload_model(model_name_control, 
                                         model_file_name_control, 
                                         framework=wallaroo.framework.Framework.ONNX)
                                         .configure(tensor_fields=["tensor"])
                        )

mainpipeline = wl.build_pipeline(main_pipeline_name)
# in case this pipeline was run before
mainpipeline.clear()
mainpipeline.add_model_step(housing_model_control)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

mainpipeline.deploy(deployment_config=deploy_config)

dataframe_start = datetime.datetime.now()

normal_input = pd.DataFrame.from_records({"tensor": [
            [
                4.0, 
                2.5, 
                2900.0, 
                5505.0, 
                2.0, 
                0.0, 
                0.0, 
                3.0, 
                8.0, 
                2900.0, 
                0.0, 
                47.6063, 
                -122.02, 
                2970.0, 
                5251.0, 
                12.0, 
                0.0, 
                0.0
            ]
        ]
    }
)
result = mainpipeline.infer(normal_input)
display(result)

large_house_input = pd.DataFrame.from_records(
    {
        'tensor': [
            [
                4.0, 
                3.0, 
                3710.0, 
                20000.0, 
                2.0, 
                0.0, 
                2.0, 
                5.0, 
                10.0, 
                2760.0, 
                950.0, 
                47.6696, 
                -122.261, 
                3970.0, 
                20000.0, 
                79.0, 
                0.0, 
                0.0
            ]
        ]
    }
)
large_house_result = mainpipeline.infer(large_house_input)
display(large_house_result)

import time
time.sleep(10)
dataframe_end = datetime.datetime.now()

batch_inferences = mainpipeline.infer_from_file('./data/xtest-1k.arrow')

large_inference_result = batch_inferences.to_pandas()
display(large_inference_result.head(20))

# pipeline log retrieval - reverse chronological order

regular_logs = mainpipeline.logs()

display("Standard Logs")
display(len(regular_logs))
display(regular_logs)

# Display metadata

metadatalogs = mainpipeline.logs(dataset=["time", "out.variable", "metadata"])
display("Metadata Logs")
# Only showing the pipeline version for space reasons
display(metadatalogs.loc[:, ["time", "out.variable", "metadata.pipeline_version"]])

# Display logs restricted by date and limit 

display("Logs restricted by date")
arrow_logs = mainpipeline.logs(start_datetime=dataframe_start, end_datetime=dataframe_end, limit=50)

display(len(arrow_logs))
display(arrow_logs)

# # pipeline log retrieval limited to arrow tables
display(mainpipeline.logs(arrow=True))

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

'Standard Logs'

100

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

'Metadata Logs'

'Logs restricted by date'

2

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

pyarrow.Table
time: timestamp[ms]
in.tensor: list<item: float>
  child 0, item: float
out.variable: list<inner: float not null> not null
  child 0, inner: float not null
anomaly.count: uint32 not null
----
time: [[2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,...,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671,2025-02-06 17:25:16.671]]
in.tensor: [[[3,2,2005,7000,1,...,1750,4500,34,0,0],[3,1.75,2910,37461,1,...,2520,18295,47,0,0],...,[4,2.25,4470,60373,2,...,3210,40450,26,0,0],[3,1,1150,3000,1,...,1460,3200,108,0,0]]]
out.variable: [[[581003],[706823.56],...,[1208638],[448627.72]]]
anomaly.count: [[0,0,0,0,0,...,0,0,0,0,0]]

# Display metadata

metadatalogs = mainpipeline.logs(dataset=['time', "out","metadata"])
display("Metadata Logs")
display(metadatalogs.loc[:, ['time', 'out.variable', 'metadata.last_model']])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

'Metadata Logs'

logs = mainpipeline.logs(limit=10000)
display(logs)

Pipeline log schema has changed over the logs requested 1000 newest records retrieved successfully, newest record seen was at <datetime>. Please request additional records separately

# Save the DataFrame version of the log file

mainpipeline.export_logs()
display(os.listdir('./logs'))

mainpipeline.export_logs(arrow=True)
display(os.listdir('./logs'))

Warning: There are more logs available. Please set a larger limit to export more data.

['logpipeline-1.arrow',
 'logpipeline-test-2.json',
 'logpipeline-1.json',
 'logpipeline-test-1.arrow',
 'logpipeline-test-2.arrow',
 'logpipeline-test-1.json']

Warning: There are more logs available. Please set a larger limit to export more data.

['logpipeline-1.arrow',
 'logpipeline-test-2.json',
 'logpipeline-1.json',
 'logpipeline-test-1.arrow',
 'logpipeline-test-2.arrow',
 'logpipeline-test-1.json']

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# Standard log retrieval

# get the workspace id
workspace_id = workspace.id()

data = {
    'pipeline_name': mainpipeline.name(),
    'workspace_id': workspace_id
}

response = requests.post(url, headers=headers, json=data)
standard_logs = pd.DataFrame.from_records(response.json())

display(len(standard_logs))
display(standard_logs.head(5).loc[:, ["time", "in", "out"]])
cursor = response.headers['x-iteration-cursor']

2

# Get next page of results as an arrow table

# retrieve the authorization token
headers = wl.auth.auth_header()
headers['Accept']="application/vnd.apache.arrow.file"

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# Standard log retrieval

data = {
    'pipeline_name': mainpipeline.name(),
    'workspace_id': workspace_id,
    'cursor': cursor
}

response = requests.post(url, headers=headers, json=data)

# Arrow table is retrieved 
with pa.ipc.open_file(response.content) as reader:
    arrow_table = reader.read_all()

# convert to Polars DataFrame and display the first 5 rows
display(arrow_table.to_pandas().head(5).loc[:,["time", "out"]])

# Upload the challenger models

model_name_challenger01 = 'logcontrolchallenger01'
model_file_name_challenger01 = './models/xgb_model.onnx'

model_name_challenger02 = 'logcontrolchallenger02'
model_file_name_challenger02 = './models/gbr_model.onnx'

housing_model_challenger01 = (wl.upload_model(model_name_challenger01, 
                                              model_file_name_challenger01, 
                                              framework=wallaroo.framework.Framework.ONNX)
                                              .configure(tensor_fields=["tensor"])
                            )
housing_model_challenger02 = (wl.upload_model(model_name_challenger02, 
                                              model_file_name_challenger02, 
                                              framework=wallaroo.framework.Framework.ONNX)
                                              .configure(tensor_fields=["tensor"])
                            )

# Undeploy the pipeline
mainpipeline.undeploy()

mainpipeline.clear()

# Add the new shadow deploy step with our challenger models
mainpipeline.add_shadow_deploy(housing_model_control, [housing_model_challenger01, housing_model_challenger02])

# Deploy the pipeline with the new shadow step
deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

mainpipeline.deploy(deployment_config=deploy_config, wait_for_status=False)

Deployment initiated for logpipeline-test. Please check pipeline status.

# wait for the pipeline status = Running
import time
time.sleep(15)

while mainpipeline.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")
    mainpipeline.status()['status']
mainpipeline.status()['status']

'Running'

# setting time period to ensure logs are different
#time.sleep(30)
shadow_date_start = datetime.datetime.now(datetime.timezone.utc)

shadow_result = mainpipeline.infer_from_file('./data/xtest-1k.arrow')

shadow_outputs =  shadow_result.to_pandas()
display(shadow_outputs.loc[0:20,['out.variable','out_logcontrolchallenger01.variable','out_logcontrolchallenger02.variable']])

#time.sleep(30)
shadow_date_end = datetime.datetime.now(datetime.timezone.utc)

# display logs with shadow deployed steps

display(mainpipeline.logs(start_datetime=shadow_date_start, end_datetime=shadow_date_end).loc[:, ["time", "out.variable", "out_logcontrolchallenger01.variable", "out_logcontrolchallenger02.variable"]])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

# display logs with shadow deployed steps

display(mainpipeline.logs(start_datetime=shadow_date_start, end_datetime=shadow_date_end).loc[:, ["time", 
                                                                                                  "out.variable", 
                                                                                                  "out_logcontrolchallenger01.variable", 
                                                                                                  "out_logcontrolchallenger02.variable"
                                                                                                  ]
                                                                                        ])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

metadatalogs = mainpipeline.logs(dataset=["time",
                                          "out_logcontrolchallenger01.variable", 
                                          "out_logcontrolchallenger02.variable", 
                                          "metadata",
                                          'anomaly.count'
                                          ],
                                start_datetime=shadow_date_start, 
                                end_datetime=shadow_date_end
                                )

display(metadatalogs.loc[:, ['out_logcontrolchallenger01.variable',	
                             'out_logcontrolchallenger02.variable', 
                             'metadata.elapsed',
                             'anomaly.count'
                             ]
                        ])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

# Save shadow deployed log files as pandas DataFrame

mainpipeline.export_logs(directory="shadow", file_prefix="shadowdeploylogs")
display(os.listdir('./shadow'))

Warning: There are more logs available. Please set a larger limit to export more data.

['shadowdeploylogs-1.json', 'shadowdeploylogs-2.json']

# Retrieve logs from specific date/time to only get the two DataFrame input inferences in ascending format

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# Standard log retrieval

data = {
    'pipeline_name': mainpipeline.name(),
    'workspace_id': workspace_id,
    'order': 'asc',
    'start_time': f'{shadow_date_start.isoformat()}',
    'end_time': f'{shadow_date_end.isoformat()}'
}

response = requests.post(url, headers=headers, json=data)

standard_logs = pd.DataFrame.from_records(response.json())

display(standard_logs.head(5).loc[:, ["time", "out", "out_logcontrolchallenger01", "out_logcontrolchallenger02"]])

mainpipeline.undeploy()

# remove the shadow deploy steps
mainpipeline.clear()

# Add the a/b test step to the pipeline
mainpipeline.add_random_split([(1, housing_model_control), (1, housing_model_challenger01), (1, housing_model_challenger02)], "session_id")

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

mainpipeline.deploy(deployment_config=deploy_config, wait_for_status=False)

Deployment initiated for logpipeline-test. Please check pipeline status.

# wait for the pipeline status = Running
import time
time.sleep(15)

while mainpipeline.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")
    mainpipeline.status()['status']
mainpipeline.status()['status']

'Running'

# Perform sample inferences of 20 rows and display the results
ab_date_start = datetime.datetime.now(datetime.timezone.utc)
abtesting_inputs = pd.read_json('./data/xtest-1k.df.json')

for index, row in abtesting_inputs.sample(20).iterrows():
    display(mainpipeline.infer(row.to_frame('tensor').reset_index()).loc[:,["out._model_split", "out.variable"]])

ab_date_end = datetime.datetime.now(datetime.timezone.utc)

## Get the logs with the a/b testing information

metadatalogs = mainpipeline.logs(dataset=["time",
                                          "out", 
                                          "metadata"
                                          ]
                                )

display(metadatalogs.loc[:, ['out.variable', 'metadata.last_model']])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

# Save a/b testing log files as DataFrame

mainpipeline.export_logs(directory="abtesting", 
                         file_prefix="abtests", 
                         start_datetime=ab_date_start, 
                         end_datetime=ab_date_end)
display(os.listdir('./abtesting'))

['abtests-1.json']

# Save a/b testing log files as DataFrame

mainpipeline.export_logs(directory="abtesting-metadata", 
                         file_prefix="abtests", 
                         start_datetime=ab_date_start, 
                         end_datetime=ab_date_end,
                         dataset=["time", "out", "metadata"])
display(os.listdir('./abtesting-metadata'))

['abtests-1.json']

# Retrieve logs from specific date/time to only get the two DataFrame input inferences in ascending format

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{wl.api_endpoint}/v1/api/pipelines/get_logs"

# Standard log retrieval

data = {
    'pipeline_name': mainpipeline.name(),
    'workspace_id': workspace_id,
    'order': 'asc',
    'start_time': f'{ab_date_start.isoformat()}',
    'end_time': f'{ab_date_end.isoformat()}'
}

response = requests.post(url, headers=headers, json=data)

standard_logs = pd.DataFrame.from_records(response.json())

display(standard_logs.head(5))

pipeline.add_validations(
    validation_name_01 = polars.col(in|out.{column_name}) EXPRESSION,
    validation_name_02 = polars.col(in|out.{column_name}) EXPRESSION
    ...{additional rules}
)

sample_pipeline = wallaroo.client.build_pipeline("sample-pipeline")
sample_pipeline.add_model_step(model)

# add the validation
sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9,
    )

# deploy the pipeline
sample_pipeline.deploy()

# sample inference
display(sample_pipeline.infer_from_file("dev_high_fraud.json", data_format='pandas-records'))

import polars as pl

mainpipeline.undeploy()
newpipeline = wl.build_pipeline("logpipeline-anomaly-example")
newpipeline.clear()
newpipeline.add_model_step(housing_model_control)
newpipeline.add_validations(
    too_high=pl.col("out.variable").list.get(0) > 1000000.0
)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

newpipeline.deploy(deployment_config=deploy_config, wait_for_status=False)

Deployment initiated for logpipeline-anomaly-example. Please check pipeline status.

# wait for the pipeline status = Running
import time
time.sleep(15)

while newpipeline.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")
    newpipeline.status()['status']
newpipeline.status()['status']

'Running'

# sample inferences
import datetime
import time
import pytz

inference_start = datetime.datetime.now(datetime.timezone.utc)

# adding sleep to ensure log distinction
time.sleep(15)

results = newpipeline.infer_from_file('./data/test-1000.df.json')

inference_end = datetime.datetime.now(datetime.timezone.utc)

# first 20 results
display(results.head(20))

# only results that trigger the anomaly too_high
results.loc[results['anomaly.too_high'] == True]

anomaly_logs =  newpipeline.logs(limit=1000)
display(anomaly_logs.loc[anomaly_logs['anomaly.too_high'] == True])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

mainpipeline.undeploy()