Implementation:Datajuicer Data juicer DAGExecutionStrategies

Knowledge Sources	Datajuicer_Data_juicer
Domains	DAG Execution, Pipeline Orchestration, Partitioning
Last Updated	2026-02-14 16:00 GMT

Overview

Implements the strategy pattern for generating DAG node structures from pipeline operations, supporting both non-partitioned (sequential) and partitioned (scatter-gather) execution models.

Description

This module is a core architectural component that decouples DAG topology generation from execution logic. It contains the following key classes:

Enums and State Machine:

• DAGNodeType -- Enum with values OPERATION, PARTITION_OPERATION, SCATTER_GATHER.
• DAGNodeStatusTransition -- Validates status transitions: PENDING->RUNNING, RUNNING->COMPLETED/FAILED, FAILED->RUNNING (retry). COMPLETED is terminal.

Node Identity:

• NodeID -- Utility for creating and parsing standardized node IDs in formats: "op_{idx}_{name}", "op_{idx}_{name}_partition_{pid}", "sg_{idx}_{name}".
• ScatterGatherNode -- Dataclass representing convergence points where partitions reconverge for global operations.

Strategy Classes:

• DAGExecutionStrategy (ABC) -- Abstract base defining generate_dag_nodes(), get_dag_node_id(), build_dependencies(), can_execute_node(), and validate_dag() (cycle detection via DFS).
• NonPartitionedDAGStrategy -- Creates simple sequential dependency chains for default and Ray executors.
• PartitionedDAGStrategy -- Generates partition-specific operation nodes plus scatter-gather convergence nodes. Scatter-gather nodes depend on ALL partitions from the previous operation; post-scatter partition ops depend on the scatter-gather node.

Helper Function:

• is_global_operation() -- Detects operations requiring full-dataset access (deduplicators, global sorts) via explicit flag, base class inheritance, or name pattern matching.

Usage

Used internally by the executor framework to convert a linear sequence of operators into a DAG structure appropriate for the chosen execution mode (default, ray, ray_partitioned).

Code Reference

Source Location

• Repository: Datajuicer_Data_juicer
• File: data_juicer/core/executor/dag_execution_strategies.py
• Lines: 1-471

Signature

class DAGNodeType(Enum):
    OPERATION = "operation"
    PARTITION_OPERATION = "partition_operation"
    SCATTER_GATHER = "scatter_gather"

class DAGNodeStatusTransition:
    @classmethod
    def is_valid(cls, from_status, to_status) -> bool: ...
    @classmethod
    def validate_and_log(cls, node_id, from_status, to_status) -> bool: ...

class DAGExecutionStrategy(ABC):
    @abstractmethod
    def generate_dag_nodes(self, operations: List, **kwargs) -> Dict[str, Any]: ...
    @abstractmethod
    def build_dependencies(self, nodes, operations, **kwargs) -> None: ...
    def validate_dag(self, nodes) -> bool: ...

class NonPartitionedDAGStrategy(DAGExecutionStrategy): ...
class PartitionedDAGStrategy(DAGExecutionStrategy):
    def __init__(self, num_partitions: int): ...

def is_global_operation(operation) -> bool: ...

Import

from data_juicer.core.executor.dag_execution_strategies import (
    DAGNodeType,
    DAGNodeStatusTransition,
    DAGExecutionStrategy,
    NonPartitionedDAGStrategy,
    PartitionedDAGStrategy,
    NodeID,
    ScatterGatherNode,
    is_global_operation,
)

I/O Contract

Inputs

Name	Type	Required	Description
operations	List[Operator]	Yes	Ordered list of pipeline operators to convert into DAG nodes
num_partitions	int	Yes (Partitioned)	Number of data partitions for PartitionedDAGStrategy
convergence_points	List[int]	No	Indices of operations requiring scatter-gather convergence

Outputs

Name	Type	Description
nodes	Dict[str, Any]	Dictionary mapping node_id to node metadata (status, dependencies, operation info)
is_valid	bool	Whether the generated DAG is acyclic

Usage Examples

from data_juicer.core.executor.dag_execution_strategies import (
    NonPartitionedDAGStrategy,
    PartitionedDAGStrategy,
    NodeID,
)

# Non-partitioned execution (sequential)
strategy = NonPartitionedDAGStrategy()
nodes = strategy.generate_dag_nodes(operations)
strategy.build_dependencies(nodes, operations)
assert strategy.validate_dag(nodes)

# Partitioned execution with scatter-gather
strategy = PartitionedDAGStrategy(num_partitions=4)
convergence_points = [i for i, op in enumerate(operations) if is_global_operation(op)]
nodes = strategy.generate_dag_nodes(operations, convergence_points=convergence_points)
strategy.build_dependencies(nodes, operations, convergence_points=convergence_points)

# Parse a node ID
info = NodeID.parse("op_001_mapper_partition_0")
# {'type': DAGNodeType.PARTITION_OPERATION, 'operation_index': 0,
#  'operation_name': 'mapper', 'partition_id': 0}

Related Pages

• data_juicer/core/executor/pipeline_dag.py (DAGNodeStatus enum)
• Datajuicer_Data_juicer_RayExecutor_Run
• Datajuicer_Data_juicer_PartitionSizeOptimizer_Calculate