Data Alignment

Function Description

Before understanding data alignment, you need to learn about data pairs.

Figure 1 Data pair diagram

• Model X is defined as the earliest data source. There are two output queues. Multiple groups of data [A1',B1'], [A2',B2'], and [A3',B3'] are output. Each group of data is associated.
• Model A and Model B are data processing nodes deployed with multiple instances across multiple devices. Model A processes data flow A', while Model B processes data flow B'. Multi-instance deployment causes processed data to become out-of-order, so the output order no longer matches that of the original outputs of Model X.
• Model C is the data aggregation node. It reads data from data flows A and B group by group for processing, and only matching pairs such as [A1, B1], [A2, B2], and [A3, B3] can be processed normally.

Data alignment means that UDF nodes require input data to be provided in matching data pairs.

Based on the data flow shown in the diagram and the framework's default receiving logic without data pairing, Model C receives data in the order [A3, B2], [A2, B1], and [A1, B3]. This disordered data cannot be processed directly. After receiving [A3, B2], the UDF node must cache the data and wait for [A2, B1] to process the valid pair [A2, B2]. It can only process [A1, B1] and [A3, B3] after receiving [A1, B3].

Constraints

Data alignment of the framework depends on transId and dataLabel. If one piece of data is distributed to different instances, differences in processing timing will make data flows reach the next node out of order, so data pairing cannot be guaranteed.

How to Use

1
2

dag = df.FlowGraph([flow_node_out])
dag.set_inputs_align_attrs(align_max_cache_num=256, align_timeout=60 * 1000, dropout_when_not_align=True) # A maximum of 256 groups of data can be cached at a timeout period of 1 minute. If the data is not aligned within the timeout period, the data is discarded.