Batching processes multiple inputs together instead of handling each input independently. In production ML, it improves the usefulness of comparisons and can improve serving throughput, but batch size and timing must match the workload. section 6.2.3 section 10.2
Common production roles include:
- Microbatch API requests before running data-drift tests; comparing a reference dataset with one request produces inaccurate results. section 6.2.3
- Maintain a moving window of recent requests, trigger a report when the configured size is reached, and evict older rows. section 6.2.3
- Feed training data through PyTorch DataLoader objects using configured training and testing batch sizes. section 9.3.2
- Let BentoML adaptively batch incoming serving requests to improve throughput and resource utilization. section 10.2
A 100-row monitoring window
Configure a monitoring service with a window size of 100. Append incoming feature rows until the window reaches 100 samples, generate a drift report against the reference dataset, then keep the most recent 100 samples as later requests arrive. A report can therefore represent a snapshot of recent API traffic rather than a single prediction. section 6.2.3
Listing 6.10 MonitoringService to run drift reports periodically
class MonitoringService:
def __init__(
self,
report: Report,
reference: pandas.DataFrame,
workspace: Workspace,
project_id: str,
window_size: int,
):
self.window_size = int(window_size)
self.report = report
self.new_rows = 0
self.reference = reference
self.workspace = workspace
self.project_id = project_id
self.current = pandas.DataFrame()
def iterate(self, new_rows: pandas.DataFrame):
rows_count = new_rows.shape[0]
self.current = pandas.concat(
[self.current, new_rows],
ignore_index=True
) #1
self.new_rows += rows_count
current_size = self.current.shape[0]
if current_size > self.window_size: #2
self.current = self.current.iloc[-self.window_size :]
self.current.reset_index(drop=True, inplace=True)
if current_size < self.window_size: #3
logger.info(
f"Not enough data for measurement: {current_size} "
f"of {self.window_size}. Waiting more data"
)
return
self.report.timestamp = datetime.datetime.now()
logger.info("Running report")
self.report.run(
reference_data=self.reference,
current_data=self.current,
) #4
self.workspace.add_report(
project_id=self.project_id,
report=self.report
) #5
Inspect the MonitoringService implementation for appending new rows, counting arrivals, enforcing the window limit, and triggering periodic reports. section 6.2.3
