hyped.core.executor module¶

Manages the execution of data flow graphs.

This module provides the DataFlowGraphExecutor class, which is responsible for executing a data flow graph. It manages the execution state, ensures that data processors are executed in the correct order, and collects the results.

class hyped.core.executor.DataFlowGraphExecutor(graph: DataFlowGraph, collect: ConcreteReference, aggregation_manager: None | DataAggregationManager)[source]¶

Bases: object

Executes a data flow graph.

This class provides the low-level functionality for executing a data flow graph, managing the execution of each node and collecting results.

build_run_context(node_id: str, index: list[int], rank: int, state: None | ExecutionState = None) → RunContext[source]¶

Build the run context for a specified node.

Parameters:

node_id (NodeId) – The id of the node.
index (IndexList) – The index list of the current batch.
rank (int) – The multiprocessing rank.
state (None | ExecutionState) – The execution state if available.

Returns:

The context for the node execution.

Return type:

RunContext

async execute(batch: Array, index: list[int], rank: int) → Array[source]¶

Execute the entire data flow graph.

Parameters:

batch (pa.Array) – The initial batch of data.
index (IndexList) – The index of the batch.
rank (Rank) – The rank of the process in a multiprocessing setting.

Returns:

The final collected batch of data.

Return type:

pa.Array

async execute_node(node_id: str, state: ExecutionState) → None[source]¶

Execute a single node in the data flow graph.

Parameters:

node_id (NodeId) – The ID of the node to execute.
state (ExecutionState) – The current execution state.

Raises:

NodeExecutionError – If the node raises an exception during execution.
AssertionError – If the output batch size doesn’t match the input batch size.

init_run_session(rank: int) → None[source]¶

Initialize the RunSession.

Parameters:: rank (Rank) – The rank of the process in a multiprocessing setting.

reset_run_session() → None[source]¶: Reset the run session.

run(batch: Table, index: list[int], rank: int | None = None) → Table[source]¶

Run the executor for a given batch.

Parameters:

batch (pa.Table) – The initial batch of data.
index (IndexList) – The index of the batch.
rank (Rank | None) – The rank of the process in a multiprocessing setting.

Returns:

The final collected batch of data.

Return type:

pa.Table

class hyped.core.executor.ExecutionState(graph: DataFlowGraph, batch: Array, index: list[int], rank: int)[source]¶

Bases: object

Tracks the state during the execution of a data flow graph.

This class is used internally to manage the state of data processing as the data flow graph is executed, keeping track of outputs, indexes, and node readiness.

capture_output(node_id: str, output: Array) → None[source]¶

Capture the output of a node.

Parameters:

node_id (NodeId) – The ID of the node producing the output.
output (pa.Array) – The output batch of data.

Raises:

AssertionError – If the node is already set

collect_inputs(node_id: str) → tuple[dict[str, Array], list[int]][source]¶

Collect inputs for a given node.

Parameters:

node_id (NodeId) – The ID of the node for which to collect inputs.

Returns:

The collected inputs to the processor: and the corresponding index

Return type:

tuple[dict[str, pa.Array], IndexList]

collect_value(ref: ConcreteReference) → Array[source]¶

Collect the values requested by the feature reference.

Parameters:: ref (ConcreteReference) – The feature reference indicating which values to collect.
Returns:: The collected pyarrow array of data.
Return type:: pa.Array
Raises:: AssertionError – If the feature reference does not contain expected feature types.

register_partition_trace(node_id: str, trace_index: list[int]) → None[source]¶

This method registers the trace indices and index mappings for a node’s output partition transition in the partition graph. It ensures that there is a valid edge between the source and target partitions and then stores the trace and index information accordingly.

Parameters:

node_id (NodeId) – The identifier of the node for which to register the partition trace.
trace_index (TraceIndexList) – The trace indices representing how to transition between the source and target partitions.

Raises:

AssertionError – If there is no edge between the source and target partitions in the partition graph.
AssertionError – If the source partition is not registered yet.

async wait_for(node_id: str) → None[source]¶

Wait until the specified node finished its execution.

Parameters:: node_id (str) – The ID of the node to wait for.

async wait_for_partition_registered(partition_id: str) → None[source]¶

Wait until a partition is registered.

Parameters:: partition_id (PartitionId) – The ID of the partition to wait for.

class hyped.core.executor.LazyDataFlowGraphExecutor(graph: DataFlowGraph, collect: ConcreteReference, input_proxy: MappingProxyType)[source]¶

Bases: Mapping, DataFlowGraphExecutor

A lazy executor for a data flow graph.

This class extends the DataFlowGraphExecutor to compute outputs only when requested and when the inputs have changed. It implements a mapping interface to provide read-only access to the output values of the data flow.

The execution is triggered lazily upon accessing an output feature, ensuring that the computation is performed only when necessary. The inputs are cached and compared to avoid redundant computations.

keys() → Iterable[str][source]¶

Get the keys of the output features.

Returns:: An iterable of the output feature keys.
Return type:: Iterable[Hashable]

exception hyped.core.executor.NodeExecutionError(node_id: str, node_obj: BaseNode, exception: Exception)[source]¶

Bases: Exception

Custom exception raised when a node execution fails in the data flow graph.