dataflow.CountBatch

Applicability

Product	Supported
Atlas A3 training product/Atlas A3 inference product	√
Atlas A2 training product/Atlas A2 inference product	√
Atlas 200I/500 A2 inference product	x
Atlas inference product	x
Atlas training product	x

Function Description

CountBatch is used to combine multiple data records into batches by batch_size based on the UDF as compute ProcessPoints. This function applies to asynchronous DataFlow scenarios. The details are as follows:

If no data is input for a long time, you can use the CountBatch function to set the timeout period. If padding is not set, the existing data will be sent to compute ProcessPoints for processing after timeout.
After the timeout period is set, if the data is less than batch_size, you can set the padding attribute by using the CountBatch function. The compute ProcessPoint fills the data to batch_size based on the padding setting and then outputs the data.

Prototype

CountBatch(batch_size=0, slide_stride=0, timeout=0, padding=False)

Parameters

Parameter	Data Type	Description
batch_size	int64_t	Size of a batch to be combined.
timeout	int64_t	This parameter takes effect only when batch_size is set. Wait time for combining batches, in ms. The value range is [0, 4294967295). The default value is 0, indicating that the group waits until the batch is full.
padding	Bool	This parameter takes effect only when batch_size and timeout are set. Whether to perform padding when the batch is insufficient. The default value is false, indicating that padding is not performed.
slide_stride	int64_t	This parameter takes effect only when batch_size is set. Sliding window stride. The value range is [0, batch_size]. If the value is greater than 0 but less than batch_size, it indicates that the sliding window is enabled for combining batches. If the value is not set or set to 0 or batch_size, batches are combined based on the mode in which the sliding window stride is not set. If the value is greater than batch_size, an error is reported.

Returns

None is returned in normal scenarios.

TypeError is returned when the parameter type is incorrect.

Examples

import dataflow as df
# Set each attribute value in count_batch as required and directly input the value through the construction method.
count_batch = df.CountBatch(batch_size=300, slide_stride=5,timeout=10,padding=300)
# Create and set the value of count_batch.
count_batch = df.CountBatch()
count_batch.batch_size = 300
# Use the map_input API of FlowNode.
df.FlowNode(...).map_input(..., [count_batch])

Constraints

Currently, the CountBatch feature cannot be used for load sharing. Therefore, if the 2P environment is used, you need to add {"ge.exec.logicalDeviceClusterDeployMode", "SINGLE"}, {"ge.exec.logicalDeviceId", "[0:0]"} during dataflow.init initialization.

Parent topic: Graph Construction APIs