DynamicBatcher

Batches records from many concurrent producers and runs them through a single async processing function, maximizing resource utilization.

Attributes

Attribute	Type	Description
stats	[BatchStats](batchstats.md?sid=flyte_extras__dynamic_batcher_batchstats)	Current BatchStats snapshot providing metrics on throughput, latency, and resource utilization.
is_running	boolean	Whether the aggregation and processing loops are active.

---

Constructor

Signature

def DynamicBatcher(
        self,
        process_fn: ProcessFn[RecordT, ResultT],
        *,
        cost_estimator: CostEstimatorFn[RecordT] | None = None,
        target_batch_cost: int = 32_000,
        max_batch_size: int = 256,
        min_batch_size: int = 1,
        batch_timeout_s: float = 0.05,
        max_queue_size: int = 5_000,
        prefetch_batches: int = 2,
        default_cost: int = 1,
    ): ...

Parameters

Name	Type	Description
process_fn	ProcessFn[RecordT, ResultT]	An async function that processes a list of records and returns a list of results in the same order.
cost_estimator	`CostEstimatorFn[RecordT]	None` = None
target_batch_cost	int = 32000	The cost budget per batch; the aggregator fills batches up to this limit.
max_batch_size	int = 256	The maximum number of records allowed in a single batch.
min_batch_size	int = 1	The minimum number of records required before dispatching a batch, unless a timeout occurs.
batch_timeout_s	float = 0.05	Maximum seconds to wait for a full batch before dispatching.
max_queue_size	int = 5000	The maximum size of the internal submission queue for backpressure.
prefetch_batches	int = 2	The number of pre-assembled batches to buffer between aggregation and processing.
default_cost	int = 1	The fallback cost assigned to a record if no estimator is available.

---

Methods

---

stats()

@property
def stats(self) -> BatchStats: ...

Current BatchStats snapshot.

Returns

Type	Description
[BatchStats](batchstats.md?sid=flyte_extras__dynamic_batcher_batchstats)	An object containing metrics such as total completed records, utilization, and average batch size.

---

is_running()

@property
def is_running(self) -> bool: ...

Whether the aggregation and processing loops are active.

Returns

Type	Description
bool	True if the batcher is currently processing or ready to process records, False otherwise.

---

start()

async def start(self) -> None: ...

Start the aggregation and processing loops.

Returns

Type	Description
None

---

stop()

async def stop(self) -> None: ...

Graceful shutdown: process all enqueued work, then stop.

Returns

Type	Description
None

---

submit()

async def submit(
        self,
        record: RecordT,
        *,
        estimated_cost: int | None = None,
    ) -> asyncio.Future[ResultT]: ...

Submit a single record for batched processing.

Parameters

Name	Type	Description
record	RecordT	The input record to be processed.
estimated_cost	`int	None` = None

Returns

Type	Description
asyncio.Future[ResultT]	A future whose result is the corresponding entry from the list returned by process_fn.

---

submit_batch()

async def submit_batch(
        self,
        records: Sequence[RecordT],
        *,
        estimated_cost: Sequence[int] | None = None,
    ) -> list[asyncio.Future[ResultT]]: ...

Convenience: submit multiple records and return their futures.

Parameters

Name	Type	Description
records	Sequence[RecordT]	An iterable of input records to be processed in batches.
estimated_cost	`Sequence[int]	None` = None

Returns

Type	Description
list[asyncio.Future[ResultT]]	List of futures, one per record, which resolve when their respective batch completes.

---