Parallel Inference

The await wallaroo.pipeline.Pipeline.parallel_infer(tensor, timeout, num_parallel, retries) asynchronous method performs an inference as defined by the pipeline steps.

Parallel Inference Parameters

Parallel Inference Returns

wallaroo.pipeline.Pipeline.parallel_infer returns one of the following based on the tensor input parameter.

• If a pandas DataFrame was submitted, a pandas DataFrame is returned.
• If an Apache Arrow table was submitted, an Apache Arrow table is returned.

The following fields are returned based on the dataset and dataset_exclude parameters.

Parallel Inference Examples

Sequential Inference Example

The first example will show taking a pandas DataFrame with 25,000 rows, splitting those into 25,000 separate DataFrames with one row each, then submitting them sequentially. This simulates a situation where the data input sizes are so large they must be broken up for more efficient transmission and inferencing.

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)}")

Elapsed = 37.917126 : 1000

Parallel Infer with DataFrame Example

The following example shows using wallaroo.pipeline.Pipeline.parallel_infer with a pandas DataFrames. It is automatically split for parallel inferences by the Wallaroo SDK.

We then compare it to the List of pandas DataFrames submitted sequentially.

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)}")

Elapsed_in_parallel = 7.691442 : 1000

parallel_results.head(20)

Parallel Infer with Apache Arrow Table

The following example shows using wallaroo.pipeline.Pipeline.parallel_infer with a pandas DataFrames. It is automatically split for parallel inferences by the Wallaroo SDK.

We then compare it to the List of pandas DataFrames submitted sequentially.

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

Elapsed_in_parallel = 3.364765 : 1000

We can quickly see the speed benefits of using parallel infer requests over sequential requests, and even more speed when going from a pandas DataFrame to an Apache Arrow table.