Model Inference

import pyarrow as pa

if input_type == "json":
            parse_elapsed += results["metadata.elapsed"][0][0]
            inference_elapsed += results["metadata.elapsed"][0][1]
        elif input_type == "arrow":
            parse_elapsed += results["metadata.elapsed"][0][0].as_py()
            inference_elapsed += results["metadata.elapsed"][0][1].as_py()
        else:
            assert False, f"Bad input type {input_type}"

result_arrow = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.arrow', dataset="metadata")

# unconverted raw data from Arrow table inference

# time to parse input data

display(result_arrow["metadata.elapsed"][0][0])

# time to inference from parsed input data

display(result_arrow["metadata.elapsed"][0][1])

<pyarrow.UInt64Scalar: 1253219>

<pyarrow.UInt64Scalar: 1275320>

parse_elapsed = result_arrow["metadata.elapsed"][0][0].as_py()
display(f'Time to parse input data: {parse_elapsed}')

'Time to parse input data: 1253219'

display(f'Time to inference from parsed data: {result_arrow["metadata.elapsed"][0][0]}')

'Time to inference from parsed data: 1253219'

# no conversion needed for pandas DataFrame

result_dataframe = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.df.json', dataset=["*", "metadata"])

result_dataframe = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.df.json', dataset=["*", "metadata"])

display(f"Time to parse input data: {result_dataframe['metadata.elapsed'][0][0]}")

display(f"Time to inference from parsed data: {result_dataframe['metadata.elapsed'][0][1]}")

'Time to parse input data: 51879691'

'Time to inference from parsed data: 2310435'

result = ccfraud_pipeline.infer(ccfraud_input_1k_arrow_table)

display(result)

pyarrow.Table
time: timestamp[ms]
in.tensor: list<item: float> not null
  child 0, item: float
out.dense_1: list<inner: float not null> not null
  child 0, inner: float not null
anomaly.count: int8
----
time: [[2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,...,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562,2023-03-20 18:55:09.562]]
in.tensor: [[[-1.0603298,2.3544967,-3.5638788,5.138735,-1.2308457,...,0.038412016,1.0993439,1.2603409,-0.14662448,-1.4463212],[-1.0603298,2.3544967,-3.5638788,5.138735,-1.2308457,...,0.038412016,1.0993439,1.2603409,-0.14662448,-1.4463212],...,[0.49511018,-0.24993694,0.4553345,0.92427504,-0.36435103,...,1.1117147,-0.566654,0.12122019,0.06676402,0.6583282],[0.61188054,0.1726081,0.43105456,0.50321484,-0.27466634,...,0.30260187,0.081211455,-0.15578508,0.017189292,-0.7236631]]]
out.dense_1: [[[0.99300325],[0.99300325],...,[0.0008533001],[0.0012498498]]]
anomaly.count: [[0,0,0,0,0,...,0,0,0,0,0]]

dataframedata = pd.DataFrame({"tensor":[npArray]})

infResults = pipeline.infer(dataframedata, dataset=["*", "metadata.elapsed"])
display(infResults.loc[0]["out.2519"])

[44,
 44,
 44,
 44,
 82,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 44,
 84,
 84,
 44,
 84,
 44,
 44,
 44,
 61,
 44,
 86,
 44,
 44]

import pyarrow as pa

if input_type == "json":
            parse_elapsed += results["metadata.elapsed"][0][0]
            inference_elapsed += results["metadata.elapsed"][0][1]
        elif input_type == "arrow":
            parse_elapsed += results["metadata.elapsed"][0][0].as_py()
            inference_elapsed += results["metadata.elapsed"][0][1].as_py()
        else:
            assert False, f"Bad input type {input_type}"

result_arrow = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.arrow', dataset="metadata")

# unconverted raw data from Arrow table inference

# time to parse input data

display(result_arrow["metadata.elapsed"][0][0])

# time to inference from parsed input data

display(result_arrow["metadata.elapsed"][0][1])

<pyarrow.UInt64Scalar: 1253219>

<pyarrow.UInt64Scalar: 1275320>

parse_elapsed = result_arrow["metadata.elapsed"][0][0].as_py()
display(f'Time to parse input data: {parse_elapsed}')

'Time to parse input data: 1253219'

display(f'Time to inference from parsed data: {result_arrow["metadata.elapsed"][0][0]}')

'Time to inference from parsed data: 1253219'

# no conversion needed for pandas DataFrame

result_dataframe = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.df.json', dataset=["*", "metadata"])

result_dataframe = ccfraud_pipeline.infer_from_file('./data/cc_data_10k.df.json', dataset=["*", "metadata"])

display(f"Time to parse input data: {result_dataframe['metadata.elapsed'][0][0]}")

display(f"Time to inference from parsed data: {result_dataframe['metadata.elapsed'][0][1]}")

'Time to parse input data: 51879691'

'Time to inference from parsed data: 2310435'

async def run_one_batch(file):
    with pa.ipc.open_file(file) as source:
        table = source.read_all() # to get pyarrow table

    # ... potentially do other processing here or request data from a feature store ...
    
    result_async_batch = await aloha_pipeline.async_infer(tensor=table,
                                                      async_client=async_client,
                                                      retries=2)
    return result_async_batch

# Process three batches in parallel. Inference on the first will start while
# the second file is read, and so on. Multiple inferences may be running at
# the same time if the engine replicas are > 1.
tasks = []
for file in ["./data/data_25k.arrow", "./data/data_1k.arrow", "./data/data_25k.arrow"]:
     tasks.append(run_one_batch(file))

# Wait for all three batches to finish.
results = await asyncio.gather(*tasks, return_exceptions=True)

for result in results:
    print(f"Number of rows for task: {result.num_rows}")
    display(result)

Number of rows for task: 24954
pyarrow.Table
time: timestamp[ms]
in.text_input: list<item: float>
  child 0, item: float
out.banjori: list<inner: float not null> not null
  child 0, inner: float not null
out.corebot: list<inner: float not null> not null
  child 0, inner: float not null
out.cryptolocker: list<inner: float not null> not null
  child 0, inner: float not null
out.dircrypt: list<inner: float not null> not null
  child 0, inner: float not null
out.gozi: list<inner: float not null> not null
  child 0, inner: float not null
out.kraken: list<inner: float not null> not null
  child 0, inner: float not null
out.locky: list<inner: float not null> not null
  child 0, inner: float not null
out.main: list<inner: float not null> not null
  child 0, inner: float not null
out.matsnu: list<inner: float not null> not null
  child 0, inner: float not null
out.pykspa: list<inner: float not null> not null
  child 0, inner: float not null
out.qakbot: list<inner: float not null> not null
  child 0, inner: float not null
out.ramdo: list<inner: float not null> not null
  child 0, inner: float not null
out.ramnit: list<inner: float not null> not null
  child 0, inner: float not null
out.simda: list<inner: float not null> not null
  child 0, inner: float not null
out.suppobox: list<inner: float not null> not null
  child 0, inner: float not null
anomaly.count: uint32 not null
----
time: [[2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,...,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406,2024-09-30 17:53:07.406]]
in.text_input: [[[0,0,0,0,0,...,32,30,19,26,17],[0,0,0,0,0,...,18,35,18,22,23],...,[0,0,0,0,0,...,27,28,19,33,23],[0,0,0,0,0,...,24,29,14,36,13]]]
out.banjori: [[[0.0015195814],[7.447168e-18],...,[1.342218e-14],[1.3068625e-12]]]
out.corebot: [[[0.98291475],[6.7359245e-8],...,[7.969789e-9],[1.1029467e-9]]]
out.cryptolocker: [[[0.012099549],[0.17081994],...,[0.14756858],[0.014839977]]]
out.dircrypt: [[[0.000047591115],[1.3220122e-9],...,[4.813928e-9],[2.2757316e-8]]]
out.gozi: [[[0.000020289312],[1.2758657e-24],...,[1.0541016e-22],[8.438438e-15]]]
out.kraken: [[[0.00031977257],[0.22559547],...,[0.2969691],[0.30495816]]]
out.locky: [[[0.011029262],[0.3420985],...,[0.2706808],[0.11627986]]]
out.main: [[[0.997564],[0.99999994],...,[1],[0.99999803]]]
...
Number of rows for task: 1000
pyarrow.Table
time: timestamp[ms]
in.text_input: list<item: float>
  child 0, item: float
out.banjori: list<inner: float not null> not null
  child 0, inner: float not null
out.corebot: list<inner: float not null> not null
  child 0, inner: float not null
out.cryptolocker: list<inner: float not null> not null
  child 0, inner: float not null
out.dircrypt: list<inner: float not null> not null
  child 0, inner: float not null
out.gozi: list<inner: float not null> not null
  child 0, inner: float not null
out.kraken: list<inner: float not null> not null
  child 0, inner: float not null
out.locky: list<inner: float not null> not null
  child 0, inner: float not null
out.main: list<inner: float not null> not null
  child 0, inner: float not null
out.matsnu: list<inner: float not null> not null
  child 0, inner: float not null
out.pykspa: list<inner: float not null> not null
  child 0, inner: float not null
out.qakbot: list<inner: float not null> not null
  child 0, inner: float not null
out.ramdo: list<inner: float not null> not null
  child 0, inner: float not null
out.ramnit: list<inner: float not null> not null
  child 0, inner: float not null
out.simda: list<inner: float not null> not null
  child 0, inner: float not null
out.suppobox: list<inner: float not null> not null
  child 0, inner: float not null
anomaly.count: uint32 not null
----
time: [[2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,...,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384,2024-09-30 17:53:07.384]]
in.text_input: [[[0,0,0,0,0,...,32,30,19,26,17],[0,0,0,0,0,...,29,12,36,31,12],...,[0,0,0,0,0,...,35,16,35,27,16],[0,0,0,0,0,...,24,29,14,36,13]]]
out.banjori: [[[0.0015195814],[0.00002837503],...,[0.0000056315566],[1.3068625e-12]]]
out.corebot: [[[0.98291475],[0.000012753118],...,[0.0000033642746],[1.1029468e-9]]]
out.cryptolocker: [[[0.012099549],[0.025435215],...,[0.13612255],[0.014839977]]]
out.dircrypt: [[[0.000047591115],[6.150966e-10],...,[5.6732154e-11],[2.2757316e-8]]]
out.gozi: [[[0.000020289312],[2.321774e-10],...,[2.773063e-8],[8.43847e-15]]]
out.kraken: [[[0.00031977257],[0.051351104],...,[0.0025221605],[0.30495816]]]
out.locky: [[[0.011029262],[0.022038758],...,[0.05455697],[0.116279885]]]
out.main: [[[0.997564],[0.9885122],...,[0.9998954],[0.99999803]]]
...
Number of rows for task: 24954
pyarrow.Table
time: timestamp[ms]
in.text_input: list<item: float>
  child 0, item: float
out.banjori: list<inner: float not null> not null
  child 0, inner: float not null
out.corebot: list<inner: float not null> not null
  child 0, inner: float not null
out.cryptolocker: list<inner: float not null> not null
  child 0, inner: float not null
out.dircrypt: list<inner: float not null> not null
  child 0, inner: float not null
out.gozi: list<inner: float not null> not null
  child 0, inner: float not null
out.kraken: list<inner: float not null> not null
  child 0, inner: float not null
out.locky: list<inner: float not null> not null
  child 0, inner: float not null
out.main: list<inner: float not null> not null
  child 0, inner: float not null
out.matsnu: list<inner: float not null> not null
  child 0, inner: float not null
out.pykspa: list<inner: float not null> not null
  child 0, inner: float not null
out.qakbot: list<inner: float not null> not null
  child 0, inner: float not null
out.ramdo: list<inner: float not null> not null
  child 0, inner: float not null
out.ramnit: list<inner: float not null> not null
  child 0, inner: float not null
out.simda: list<inner: float not null> not null
  child 0, inner: float not null
out.suppobox: list<inner: float not null> not null
  child 0, inner: float not null
anomaly.count: uint32 not null
----
time: [[2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,...,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395,2024-09-30 17:53:07.395]]
in.text_input: [[[0,0,0,0,0,...,32,30,19,26,17],[0,0,0,0,0,...,18,35,18,22,23],...,[0,0,0,0,0,...,27,28,19,33,23],[0,0,0,0,0,...,24,29,14,36,13]]]
out.banjori: [[[0.0015195814],[7.447168e-18],...,[1.342218e-14],[1.3068625e-12]]]
out.corebot: [[[0.98291475],[6.7359245e-8],...,[7.969789e-9],[1.1029467e-9]]]
out.cryptolocker: [[[0.012099549],[0.17081994],...,[0.14756858],[0.014839977]]]
out.dircrypt: [[[0.000047591115],[1.3220122e-9],...,[4.813928e-9],[2.2757316e-8]]]
out.gozi: [[[0.000020289312],[1.2758657e-24],...,[1.0541016e-22],[8.438438e-15]]]
out.kraken: [[[0.00031977257],[0.22559547],...,[0.2969691],[0.30495816]]]
out.locky: [[[0.011029262],[0.3420985],...,[0.2706808],[0.11627986]]]
out.main: [[[0.997564],[0.99999994],...,[1],[0.99999803]]]
...

test_data = pd.read_json("./data/data_25k.df.json")
test_list = []

for index, row in test_data.head(1000).iterrows():
    test_list.append(row.to_frame('text_input').reset_index(drop=True))

# show the first row as an example
test_list[0]

#
# Run the inference sequentially to establish a baseline
#
now = datetime.datetime.now()

results = []
for df in test_list:
    results.append(aloha_pipeline.infer(tensor=df, timeout=10))

total_sequential = datetime.datetime.now() - now

print(f"Elapsed = {total_sequential.total_seconds()} : {len(results)}")

timeout_secs=30
now = datetime.datetime.now()
parallel_results = await aloha_pipeline.parallel_infer(tensor=test_data.head(1000), 
                                                       timeout=timeout_secs, 
                                                       num_parallel=2*REPLICAS, 
                                                       retries=3)

total_parallel = datetime.datetime.now() - now
print(f"Elapsed_in_parallel = {total_parallel.total_seconds()} : {len(parallel_results)}")

parallel_results.head(20)

# convert the DataFrame to an Arrow Table

import pyarrow as pa
timeout_secs=30
now = datetime.datetime.now()

test_data_arrow_table = pa.Table.from_pandas(test_data.head(1000))

parallel_results = await aloha_pipeline.parallel_infer(tensor=test_data_arrow_table, 
                                                       timeout=timeout_secs, 
                                                       num_parallel=2*REPLICAS, 
                                                       retries=3)

total_parallel = datetime.datetime.now() - now
print(f"Elapsed_in_parallel = {total_parallel.total_seconds()} : {len(parallel_results)}")

{
    "username": "{Connecting User's Username}", 
    "password": "{Connecting User's Password}", 
    "email": "{Connecting User's Email Address}"
}

import wallaroo
from wallaroo.object import EntityNotFoundError

import pandas as pd
import os
import base64

import pyarrow as pa

import requests
from requests.auth import HTTPBasicAuth

# Used to create unique workspace and pipeline names
import string
import random

import json

# used to display dataframe information without truncating
from IPython.display import display
pd.set_option('display.max_colwidth', None)

# Retrieve the login credentials.
os.environ["WALLAROO_SDK_CREDENTIALS"] = './creds.json.example'

# wl = wallaroo.Client(auth_type="user_password")

# Client connection from local Wallaroo instance
wallarooDomain = 'wallaroo.example.com'

wl = wallaroo.Client(api_endpoint=f"https://{wallarooDOmain}", 
                    auth_type="user_password")

wallarooDomain = "wallaroo.example.com"

APIURL=f"https://{wallarooDomain}/v1/api"

# Retrieve the token
headers = wl.auth.auth_header()
display(headers)

# Retrieve the token
headers = wl.auth.auth_header()

# set Content-Type type
headers['Content-Type']='application/json'

## Retrieve the pipeline's External Inference URL

apiRequest = f"{APIURL}/v1/api/admin/get_pipeline_external_url"

data = {
    "workspace_id": workspaceId,
    "pipeline_name": pipeline_name
}

response = requests.post(apiRequest, json=data, headers=headers, verify=True).json()
deployurl = response['url']
deployurl

# Retrieve the token
headers = wl.auth.auth_header()

# set Content-Type type
headers['Content-Type']='application/json; format=pandas-records'

## Inference through external URL using dataframe

# retrieve the json data to submit
data = [
    {
        "tensor":[
            1.0678324729,
            0.2177810266,
            -1.7115145262,
            0.682285721,
            1.0138553067,
            -0.4335000013,
            0.7395859437,
            -0.2882839595,
            -0.447262688,
            0.5146124988,
            0.3791316964,
            0.5190619748,
            -0.4904593222,
            1.1656456469,
            -0.9776307444,
            -0.6322198963,
            -0.6891477694,
            0.1783317857,
            0.1397992467,
            -0.3554220649,
            0.4394217877,
            1.4588397512,
            -0.3886829615,
            0.4353492889,
            1.7420053483,
            -0.4434654615,
            -0.1515747891,
            -0.2668451725,
            -1.4549617756
        ]
    }
]

# submit the request via POST, import as pandas DataFrame
response = pd.DataFrame.from_records(
    requests.post(
        deployurl, 
        json=data, 
        headers=headers)
        .json()
    )

display(response.loc[:,["time", "out"]])

# Retrieve the token
headers = wl.auth.auth_header()

# set Content-Type type
headers['Content-Type']='application/vnd.apache.arrow.file'

# set accept as apache arrow table
headers['Accept']="application/vnd.apache.arrow.file"

# Submit arrow file
dataFile="./data/cc_data_10k.arrow"

data = open(dataFile,'rb').read()

response = requests.post(
                    deployurl, 
                    headers=headers, 
                    data=data, 
                    verify=True
                )

# 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"]])