data_juicer.ops.fused_sequential_batch_op module#

FusedSequentialBatchOp - run multiple batch-local ops in one stage.

This fused op reduces scheduler/stage overhead by executing a list of batch-local sub-operators sequentially inside one dataset map stage. Each sub-op receives the batch returned by the previous sub-op, so normal mapper and filter chains remain possible, including filters that drop rows.

class data_juicer.ops.fused_sequential_batch_op.FusedSequentialBatchOp(*args, **kwargs)[source]#

Bases: Mapper

Run multiple batch-local mapper/filter ops sequentially in one stage.

Supports two initialization modes:

  1. op_specs mode: list of {“class_name”: str, “kwargs”: dict}.

  2. fused_ops mode: list of pre-built op instances.

This class intentionally does not fan out work across threads. Its primary purpose is reducing stage overhead while preserving normal sequential semantics.

__init__(op_specs: List[Dict[str, Any]] | None = None, fused_ops: List[Any] | None = None, group_name: str = '', cleanup_columns: List[str] | None = None, *args, **kwargs)[source]#
Parameters:

  • op_specs – sub-op specs. Ray scheduling kwargs are stripped before constructing sub-ops.

  • fused_ops – already-instantiated batch-local ops.

  • group_name – human-readable label used in logs.

  • cleanup_columns – top-level columns to remove after all sub-ops finish.

process_batched(samples, rank=None)[source]#

Run sub-ops sequentially, passing each returned batch onward.