hyped package

Subpackages

Submodules

Module contents

A comprehensive framework for complex data processing pipelines.

This package provides a comprehensive framework for constructing, managing, and executing complex data processing pipelines. The framework is designed to be modular and flexible, allowing users to define data flows to handle a wide variety of data processing tasks.

class hyped.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

hyped.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