hyped.core.flow module

Defines high-level interfaces for data processing workflows.

This module provides the DataFlow class, which allows users to define and execute complex data processing workflows. The workflows are represented as directed acyclic graphs (DAGs) of data processors.

class hyped.core.flow.DataFlow(features: None | Features = None)[source]

Bases: AbstractDataFlow, Generic[T]

Data Flow.

The DataFlow class allows users to create and manage directed acyclic graphs (DAGs) of data processors, facilitating complex data transformations and processing pipelines. Users can easily define source features, build sub-flows for specific outputs, and apply these workflows to batches of data or entire HuggingFace datasets.

This class integrates various components such as the data flow graph and the executor to provide a seamless experience for processing data. It handles the internal state management, execution scheduling, and data flow dependencies to ensure efficient and accurate data processing.

U = ~U
apply(ds: Dataset, collect: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False, keep_in_memory: bool = False, load_from_cache_file: bool = True, writer_batch_size: int = 1000, num_proc: None | int = None, desc: None | str = None) → Dataset[source]
apply(ds: Dataset, collect: Feature | dict[str, Any], aggregate: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False, keep_in_memory: bool = False, load_from_cache_file: bool = True, writer_batch_size: int = 1000, num_proc: None | int = None, desc: None | str = None) → tuple[Dataset, dict[str, Any]]
apply(ds: ItDataset, collect: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False) → ItDataset
apply(ds: ItDataset, collect: Feature | dict[str, Any], aggregate: Feature | dict[str, Any], *, debug: bool = False, batch_size: int = 1000, drop_last_batch: bool = False) → tuple[ItDataset, dict[str, Any]]

Apply the data flow graph to a dataset.

This method processes a given dataset or iterable dataset using the data flow graph, executing transformations based on the collect and optional aggregate features. The behavior of the processing depends on whether the dataset is an in-memory or streamed dataset (i.e. datasets.Dataset or datasets.IterableDataset)).

Parameters:

  • ds (Dataset | IterableDataset) – The dataset to which the data flow graph will be applied.

  • collect (Feature | dict[str, Any]) – The collect feature, which defines the primary transformations applied to the dataset.

  • aggregate (None | Feature | dict[str, Any]) – An optional aggregate feature, which applies additional aggregate-level transformations. If not provided, aggregation is skipped.

  • debug (bool) – If True, executes debug nodes allowing for inspection of intermediate data. Defaults to False.

  • batch_size (int) – The number of samples to process in a batch. Defaults to 1000.

  • drop_last_batch (bool) – Whether to drop the last batch if it is smaller than the specified batch size. Defaults to False.

  • keep_in_memory (bool) – If True, the resulting dataset is kept in memory. Defaults to False. Only applies for datasets.Dataset.

  • load_from_cache_file (bool) – Whether to load the resulting dataset from cache files when possible. Defaults to True. Only applies for datasets.Dataset.

  • writer_batch_size (int) – Batch size for writing results to cache files. Defaults to 1000. Only applies for datasets.Dataset.

  • num_proc (None | int) – The number of processes to use for parallel processing. Defaults to None, which disables multiprocessing. Only applies for datasets.Dataset.

  • desc (Optional[str], optional, only for Dataset) – A description for the progress bar displayed during processing. Defaults to None. Only applies for datasets.Dataset.

Returns:

The transformed dataset matching the input dataset type and a dictionary containing the aggregation results when aggregation is applied, i.e. the aggregate input is specified.

Return type:

Dataset | ItDataset | tuple[Dataset, dict[str, Any]] | tuple[ItDataset, dict[str, Any]]

build(collect: Feature | dict[str, Any], aggregate: None | Feature | dict[str, Any] = None, *, aggregation_manager: DataAggregationManager | None = None, debug: bool = True) → ExecutableDataFlow[source]

Build an executable data flow for computing and collecting features.

This method constructs a read-only, executable representation of the data flow for a specified collect feature. Optionally, an aggregate feature can also be included for dataset-wide computations. The resulting data flow is optimized for execution.

Parameters:

  • collect (Feature | dict[str, Any]) – The feature to be computed and collected.

  • aggregate (None | Feature) – An optional feature for computing aggregated values across the dataset.

  • *

  • aggregation_manager (DataAggregationManager | None) – The data aggregation manager instance.

  • debug (bool) – If True, includes debug nodes in the built data flow, allowing for inspection of intermediate data. Defaults to True.

Returns:

An executable data flow that encapsulates the graph, collect feature, and optional aggregate feature.

Return type:

ExecutableDataFlow[T]

Raises:

  • RuntimeError – If the collect feature does not belong to the current data flow graph.

  • RuntimeError – If the aggregate feature does not belong to the current data flow graph.

collect(collect: dict[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any]) → _MappingFeature[source]
collect(collect: list[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any] | tuple[dict[str, dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | list[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | tuple[dict[str, NestedType] | list[NestedType] | tuple[NestedType] | T] | Any]) → SequenceFeature
collect(collect: Any) → Feature

Add a collect node to the data flow.

The collect() method processes a nested structure (e.g., dicts, lists, tuples) of constants and features and adds a corresponding collect node to the data flow graph. Constants are converted to constant nodes, while features are directly incorporated.

Parameters:

collect (NestedType[Any]) – A nested structure of constants, Feature objects, and Reference instances. The structure may include dictionaries, lists, and tuples, where constants are automatically added as constant nodes.

Returns:

A feature or nested structure of features representing

the collected data.

Return type:

Mapping | Sequence | Feature

Raises:

  • NotImplementedError – If an empty sequence (list or tuple) is encountered.

  • TypeError – If an unsupported type is encountered in the nested structure.

const(value: int) → Int64Feature[source]
const(value: float) → Float64Feature
const(value: bool) → BoolFeature
const(value: str) → StringFeature
const(value: dict[str, Any]) → _MappingFeature
const(value: list[Any] | tuple[Any]) → SequenceFeature
const(value: Any, feature_type: type[U]) → U

Add a constant value as a node to the data flow.

This method allows the user to add a constant value as a node in the data flow graph. The constant can either be associated with a specific feature type or have its data type inferred from the provided value.

Parameters:

  • value (Any) – The constant value to add to the data flow graph.

  • feature_type (None | type[Feature]) – The explicit feature type for the value. If provided, the data type will be validated and inferred from this type. If not provided, the data type is inferred from the value.

Returns:

The feature representation of the constant node added to the graph.

Return type:

Feature

Raises:

TypeError – If the provided feature_type is invalid or if the value cannot be validated against the feature_type.

property depth: int

Computes the total depth of the data flow graph.

The depth is defined as the maximum level of any node in the graph, where the root node has a depth of 0. This property calculates the depth by finding the maximum depth attribute among all nodes in the graph.

Returns:

The total depth of the graph.

Return type:

int

classmethod deserialize(data: str, debug: bool = True) → ExecutableDataFlow[source]

Deserializes a JSON string into an ExecutableDataFlow instance.

This method parses a JSON string into a dictionary, validates its structure, and uses the data to reconstruct the executable data flow, including references to the collection and aggregation nodes.

Parameters:

  • data (str) – The JSON string representing the serialized executable data flow.

  • debug (bool) – If True, the deserialized graph will include debug nodes present in the serialized flow. If False, debug nodes will be omitted from the deserialized graph. Defaults to True.

Returns:

The deserialized executable data flow.

Return type:

ExecutableDataFlow

Raises:

ValueError – If the input JSON string does not contain the required keys (“graph”, “collect”, and “aggregate”).

property source: T

Get the source features.

Returns:

The reference to the source features.

Return type:

T

property width: int

Computes the maximum width of the data flow graph.

The width is defined as the maximum number of nodes present at any single depth level in the graph. This property calculates the width by grouping nodes by their depth and finding the largest group.

Returns:

The maximum width of the graph.

Return type:

int

class hyped.core.flow.ExecutableDataFlow(source_annotation: Any | None, graph: DataFlowGraph, collect: ConcreteReference, aggregate: ConcreteReference | None, aggregation_manager: DataAggregationManager | None, debug: bool)[source]

Bases: AbstractDataFlow

Executable data flow.

This class validates and optimizes the input data flow graph and provides methods to execute the flow over a dataset. It manages source, collect, and aggregate features, optimizing and partitioning the graph into instance and aggregate graphs.

This class is not meant to be instantiated directly. Use the DataFlow.build() function to construct an instance, ensuring proper setup and validation of the data flow.

property aggregates: Mapping[str, Any]

Read-only view of the aggregated values computed during execution.

Returns:

A read-only mapping of aggregated values.

Return type:

MappingProxyType[str, Any]

property aggregates_feature: None | _MappingFeature

The aggregates feature of the executable data flow.

Returns:

The aggregates feature that is computed when the data flow is

executed. None in case no aggregates were specified when building the flow.

Return type:

None | MappingFeature

apply(ds: Dataset, *, batch_size: int = 1000, drop_last_batch: bool = False, keep_in_memory: bool = False, load_from_cache_file: bool = True, writer_batch_size: int = 1000, num_proc: None | int = None, desc: None | str = None) → Dataset[source]
apply(ds: ItDataset, *, batch_size: int = 1000, drop_last_batch: bool = False) → ItDataset

Apply the data flow graph to a dataset.

This method processes a given dataset or iterable dataset using the data flow graph. The behavior of the processing depends on whether the dataset is an in-memory or streamed dataset (i.e. datasets.Dataset or datasets.IterableDataset)).

If the dataset features do not fully align with the source features of the data flow graph, a fallback mechanism applies a casting operation. This ensures compatibility but may result in data loss or type adjustments. The casting adapts the dataset features to match the schema defined by the data flow graph.

Parameters:

  • ds (Dataset | IterableDataset) – The dataset to which the data flow graph will be applied.

  • batch_size (int) – The number of samples to process in a batch. Defaults to 1000.

  • drop_last_batch (bool) – Whether to drop the last batch if it is smaller than the specified batch size. Defaults to False.

  • keep_in_memory (bool) – If True, the resulting dataset is kept in memory. Defaults to False. Only applies for datasets.Dataset.

  • load_from_cache_file (bool) – Whether to load the resulting dataset from cache files when possible. Defaults to True. Only applies for datasets.Dataset.

  • writer_batch_size (int) – Batch size for writing results to cache files. Defaults to 1000. Only applies for datasets.Dataset.

  • num_proc (None | int) – The number of processes to use for parallel processing. Defaults to None, which disables multiprocessing. Only applies for datasets.Dataset.

  • desc (Optional[str], optional, only for Dataset) – A description for the progress bar displayed during processing. Defaults to None. Only applies for datasets.Dataset.

Returns:

The transformed dataset matching the input dataset type.

Return type:

Dataset | ItDataset | tuple[Dataset

attach(node: _MappingFeature) → tuple[_MappingFeature, _MappingFeature | None][source]

Attach the data flow graph to a given node.

This method attaches a data flow graph (self) to a specified node of an existing graph. Attaching the data flow graph means that the source node of the current graph is connected to the given node, effectively embedding the data flow graph into the target graph at the specified location.

Parameters:

node (MappingFeature) – The target node in the existing graph to which the data flow graph will be attached.

Returns:

A tuple containing:

  • The updated collect feature, which now references the corresponding node in the attached graph.

  • The updated aggregate feature, or None if the data flow graph does not have an aggregate feature.

Return type:

tuple[MappingFeature, MappingFeature | None]

property collect_feature: _MappingFeature

The collect feature of the executable data flow.

Returns:

The output feature that is collected by the data flow when executed.

Return type:

MappingFeature

property depth: int

Computes the total depth of the data flow graph.

The depth is defined as the maximum level of any node in the graph, where the root node has a depth of 0. This property calculates the depth by finding the maximum depth attribute among all nodes in the graph.

Returns:

The total depth of the graph.

Return type:

int

classmethod deserialize(data: str, debug: bool = True) → ExecutableDataFlow[source]

Deserializes a JSON string into an ExecutableDataFlow instance.

This method parses a JSON string into a dictionary, validates its structure, and uses the data to reconstruct the executable data flow, including references to the collection and aggregation nodes.

Parameters:

  • data (str) – The JSON string representing the serialized executable data flow.

  • debug (bool) – If True, the deserialized graph will include debug nodes present in the serialized flow. If False, debug nodes will be omitted from the deserialized graph. Defaults to True.

Returns:

The deserialized executable data flow.

Return type:

ExecutableDataFlow

Raises:

ValueError – If the input JSON string does not contain the required keys (“graph”, “collect”, and “aggregate”).

serialize(indent: None | int = None) → str[source]

Serializes the executable data flow into a JSON string.

This method creates a dictionary representation of the entire data flow, including the graph and the references for collection and aggregation nodes. The dictionary is then serialized into a JSON string.

Parameters:

indent (None | int, optional) – If provided, formats the output with the given indentation level. If None, the JSON is serialized without indentation.

Returns:

The JSON string representing the serialized executable data flow.

Return type:

str

property width: int

Computes the maximum width of the data flow graph.

The width is defined as the maximum number of nodes present at any single depth level in the graph. This property calculates the width by grouping nodes by their depth and finding the largest group.

Returns:

The maximum width of the graph.

Return type:

int

hyped.core.flow.plot_data_flow(flow: DataFlow, node_format: str | Template = "[{{ node_id[:4] }}] {% if node_type == 'SOURCE_NODE' %}Source{% else %}{{ node_object }}{% endif %}", with_edge_labels: bool = True, edge_font_size: int = 6, node_font_size: int = 6, node_size: int = 5000, arrowsize: int = 25, color_map: dict[NodeType, str] = {}, legend: bool = True, legend_fontsize: int = 6, ax: None | Axes = None) → Axes[source]

Plot a data flow graph.

Parameters:

  • flow (DataFlow) – The data flow to plot.

  • node_format (str | Template) – The jinja template used to generate node labels.

  • with_edge_labels (bool) – Whether to include labels on the edges. Defaults to True.

  • edge_font_size (int) – The font size for edge labels. Defaults to 6.

  • node_font_size (int) – The font size for node labels. Defaults to 6.

  • node_size (int) – The size of the nodes. Defaults to 5_000.

  • arrowsize (int) – The size of the arrows on the edges. Defaults to 25.

  • color_map (dict[None | type, str]) – indicate custom color scheme based on the processor type. None refers to the source node.

  • legend (bool) – Whether to add a legend of the node types to the axes. Defaults to True.

  • legend_fontsize (int) – The font size for the legend. Defaults to 6.

  • ax (Optional[plt.Axes]) – Matplotlib axes object to draw the plot on. Defaults to None.

Returns:

The Matplotlib axes object with the plot.

Return type:

plt.Axes