data_juicer.ops.op_fusion module

data_juicer.ops.op_fusion.fuse_operators(ops, probe_res=None, mapper_fusion=True, mapper_fusion_vram_limit=0.9)

Fuse the input ops list and return the fused ops list.

参数:

• ops -- the corresponding list of op objects.

• probe_res -- the probed speed for each OP from Monitor.

• mapper_fusion -- whether to fuse consecutive independent GPU Mappers into FusedSequentialBatchOp for single-stage execution. Only effective when op_fusion is true.

• mapper_fusion_vram_limit -- max aggregate GPU memory budget (fraction of one GPU) for a fused mapper group. Default 0.9.

返回:

a list of fused op objects.

data_juicer.ops.op_fusion.fuse_filter_group(original_filter_group)

Fuse single filter group and return the fused filter group.

参数:

original_filter_group -- the original filter group, including op definitions and objects.

返回:

the fused definitions and objects of the input filter group.

class data_juicer.ops.op_fusion.FusedFilter(*args, **kwargs)

基类：Filter

A fused operator for filters.

__init__(name: str, fused_filters: List)

Initialization method.

参数:

fused_filters -- a list of filters to be fused.

compute_stats_batched(samples, rank=None)

process_batched(samples)

class data_juicer.ops.op_fusion.GeneralFusedOP(*args, **kwargs)

基类：Mapper

An explicitly fused operator designed to execute multiple sequential operations (OPs) on the same batch, enabling fine-grained control over data processing.

This operator allows for the chaining of multiple data processing steps, such as mappers and filters, into a single pass. It processes each batch of samples sequentially through the defined operations, ensuring that all specified transformations are applied in order. The operator supports both mappers, which transform data, and filters, which remove or keep samples based on computed statistics. Context variables can be passed between operations if needed. The accelerator is set to 'cuda' if any of the fused operations use it. The number of processes is determined by the minimum value among all fused operations. After processing, any temporary context variables, such as those used for video containers, are cleaned up.

__init__(batch_size: int = 1, fused_op_list: List | None = None, *args, **kwargs)

Initialization.

参数:

• batch_size -- the batch size of the input samples.

• fused_op_list -- a list of OPs to be fused.

process_batched(samples, rank=None)

run(dataset, *, exporter=None, tracer=None)

data_juicer.ops.op_fusion.fuse_mapper_group(mapper_group: list, vram_limit: float = 0.9) → list

Fuse consecutive independent GPU Mappers into FusedSequentialBatchOp.

Safety rules: - All ops must be Mapper instances with num_gpus > 0 - All ops must explicitly opt in with _fused_sequential_batch_op_safe = True - Ops must be independent (disjoint declared output columns) - All ops must declare estimated_vram_fraction - Aggregate estimated VRAM should not exceed vram_limit - All ops must use the same Ray runtime environment

Returns a list with either the original ops (if not fuseable) or a single FusedSequentialBatchOp wrapping the group.

data_juicer.ops.op_fusion.fuse_consecutive_mappers(ops: list, vram_limit: float = 0.9) → list

Scan op list and fuse consecutive GPU Mapper groups.

Groups are delimited by non-Mapper ops or CPU ops. Each group of >= 2 consecutive GPU Mappers is fused into a FusedSequentialBatchOp. Single GPU Mappers pass through.