dgf.transform

dgf.transform.AutoNormalizeConfig

Configuration for automatic feature normalization for GNNs.

Attributes:

Name	Type	Description
`categorical_bytes_to_index`	`bool`	If True, categorical features with BYTES format will be converted to integer indices using `DictionaryIndexNormalizer`.
`numerical_soft_quantile`	`bool`	If True, numerical features (INTEGER or FLOAT) will be normalized using `SoftQuantileNormalizer`.
`keep_raw_features`	`Set[str]`	A list of feature names that should bypass all normalization and be included in the output graph as-is. This is useful for preserving features like unique identifiers or other metadata not intended for model input.
`ignore_features_without_stats`	`bool`	Whether to ignore features that are not in `keep_raw_features` and do not have associated statistics. If `False` (default), an error is raised. If `True`, such features are skipped.
`consume_primary_keys`	`bool`	If True, features marked as PRIMARY_ID will be normalized. If false, primary key features are skipped.
`primary_keys_num_hash_buckets`	`int`	Number of hash buckets to use when normalizing primary key features.

dgf.transform.ContainsLabelPredicate

Predicate for filtering subgraphs if they have a positive label.

Attributes:

Name	Type	Description
`nodeset_name`	`str`	Target nodeset name.
`feature_name`	`str`	The feature name we want to filter the `label` on.
`label`	`Any`	The label value we want to filter.
`aggregator`	`str`	"ANY" or "ALL" aggregator function.

dgf.transform.DictionaryIndexNormalizer

Bases: AbstractFeatureNormalizer

Normalizes features by mapping dictionary keys to their integer indices.

This normalizer is suitable for categorical features where a dictionary of unique values and their assigned indices is available in the statistics.

If the feature stats indexing is a dense [0, num_items) indexing (which should be the case), the output will be in [0, num_items+1).

This operation is the same as the "compression mapping" in GraphAI.

dgf.transform.GNNDatasetPreparator

Generates graph samples to train node prediction models.

The following transformations are applied: - Feature normalization (e.g., soft quantile, indexing) using dgf.analyse.feature_statistics_from_graphs and dgf.transform.AutoNormalizer. - Padding graphs using dgf.analyse.padding_from_graph_generator. - Merging of batches of graphs using dgf.transform.merge_graphs.

This class is intended for basic GNN pipelines. For advanced GNN pipelines, users should apply those transformations manually.

Usage example:

graph, schema = dgf.io.read_graph(DATASET_DIR)

train_dataset = dgf.transform.GNNDatasetPreparator(
    graph=raw_graph,
    schema=raw_schema,
    sampling_config=dgf.sampling.SimpleSamplingConfig(
        seed_nodeset="nodes",
        num_hops=2,
        hop_width=3,
        reverse=True,
    ),
    batch_size=10,
    drop_remainder=True,
    shuffle=True,
)

# Generate batched, normalized, padded graphs.
for graph_sample, merge_offsets in train_dataset.generate():
  print(graph_sample)
  print(merge_offsets)
  break

Attributes:

Name	Type	Description
`graph`	`Graph`	One of the graph format defined in data.Graph e.g. in-memory graph, generator of graph samples, path to graph samples.
`schema`	`GraphSchema`	The schema of the graph. The semantics of the features should be configured.
`batch_size`	`int`	The desired size of each batch of seed nodes. Set batch_size=1 to avoid batching / merging.
`sampling_plan`	`SamplingPlan`	Configuration for sampling subgraphs around the seed nodes.
`drop_remainder`	`bool`	If `True`, the last batch of seed nodes will be dropped if it contains fewer than `batch_size` elements.
`shuffle`	`bool`	If `True`, the seed nodes are shuffled before batching.
`num_samples_for_stats`	`Optional[int]`	Number of graph samples used to compute feature and graph statistics, which are required for feature normalization and padding. Larger values lead to more precise statistics but increase the time taken during the initial preparation phase. If set to `None`, statistics will be computed using one sample for each node in the seed nodeset.
`auto_normalize_config`	`AutoNormalizeConfig`	Configuration for automatic feature normalization. Defaults to `normalize_lib.AutoNormalizeConfig()`.
`verbose_preparation`	`bool`	If true, display a progress bar during the preparation stage that appears only the first time `generate` is called.
`skip_overflow_padding_error`	`bool`	If padding is set, the merging stage can fail if the "padding" is not large enought. If skip_overflow_padding_error=True, such batch is skipped. If skip_overflow_padding_error=False, and error is raised.
`format`	`Union[GraphFormat, str]`	The format of the graph. If set to AUTO, the format will be inferred from the graph object.
`seed_node_idxs`	`Optional[ndarray]`	Optional array of node indices to use as seeds for sampling. If None, all nodes in the seed nodeset are used.
`temporal_sampling`	`bool`	True if the sampling relies on timestamp features to condition the sampling.
`edgeset_timestamp_features`	`Dict[str, str]`	Optional dictionary mapping edgeset names to the feature name containing timestamp information.
`nodeset_timestamp_features`	`Dict[str, str]`	Optional dictionary mapping nodeset names to the feature name containing timestamp information.
`cache_normalized_features`	`bool`	If True, pre-compute the normalized features during the preparation stage instead of computing them on the fly in the generator. This optioncan speeds up data generation/training but increases memory consumption. This option is only available for in memory graph inputs.
`cache_normalized_features_device`	`Literal['host', 'device']`	Specifies the device ("host" for RAM or "device" for GPU/TPU) to store the cached normalized features. Caching features on the same device used for training reduces host-device communication, potentially speeding up training, but increases memory consumption on the device.

`generate`

Generates batched + normalized graph samples.

Yields:

Type	Description
`InMemoryGraph`	A tuple containing the normalized, merged, and padded graph sample, and
`Dict[str, ndarray]`	the merging node offsets.

`generate_jax`

Generates jax batched + normalized graph samples.

Yields:

Type	Description
`JaxInMemoryGraph`	A tuple containing the normalized, merged, and padded graph sample, and
`Dict[str, ndarray]`	the merging node offsets.

`generated_schema`

Returns the schema of the generated samples.

`num_nodes_in_seed_nodeset`

Number of nodes in the seed nodeset.

`prepare`

Pre-compute and pre-pare what is necessary for the generation.

Can only be called once. Called automatically the first time "generate" is called.

`prepare_from_existing_one`

Pre-compute data (same as "prepare") but with an already computed cache.

Instead of beeing recomputed, the following are grabbed from "other": - Feature statistics - Normalier - Padding - Sampling plan

Parameters:

Name	Type	Description	Default
`other`	`GNNDatasetPreparator`	Another dataset preparator from which to copy the data.	required

dgf.transform.GraphNormalizer

Applies a collection of individual AbstractFeatureNormalizer on a graph.

A graph normalier prepares features values before they can be consumed by a core GNN model.

`get_normalized_feature_names`

Gets the normalized feature names derived from a given input feature.

An original feature can be transformed into one or more new features by different normalizers. This method returns the names of all such generated features.

Parameters:

Name	Type	Description	Default
`nodeset_name`	`str`	The name of the node set containing the feature.	required
`original_feature_name`	`str`	The name of the input feature before normalization.	required

Returns:

Type	Description
`List[str]`	A list of feature names in the normalized graph that originated from
`List[str]`	`original_feature_name`.

`normalize_numpy`

Normalizes the features of the input graph.

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The input `InMemoryGraph` with raw feature values.	required

Returns:

Type	Description
`InMemoryGraph`	A new `InMemoryGraph` with normalized feature values.

`normalize_numpy_to_jax`

Normalizes graph features using numpy and returns jax arrays.

This function is equivalent, but consumes less memory than, calling "normalize" + "graph_to_jax_graph".

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The input `InMemoryGraph` with raw feature values.	required
`include_adjacencies`	`bool`	If True, the adjacency information from the input graph will be included in the output `JaxInMemoryGraph`. Otherwise, the output graph's edge sets will have `None` for adjacency.	`True`

Returns:

Type	Description
`JaxInMemoryGraph`	A new `JaxInMemoryGraph` with normalized feature values.

`normalize_tensorflow`

Normalizes the features of the input graph using tensorflow.

Parameters:

Name	Type	Description	Default
`graph`	`TFInMemoryGraph`	The input `InMemoryGraph` with raw feature values.	required

Returns:

Type	Description
`TFInMemoryGraph`	A new `InMemoryGraph` with normalized feature values.

`output_schema`

Returns the schema of the graph after normalization.

This schema reflects the changes in feature formats, semantics, and shapes resulting from the applied normalizations.

`tensorflow_resources`

Returns all the tf resources of all the operations.

dgf.transform.GraphNormalizerConfig

Raw information of a GraphNormalizer for easy serialization.

`nice_print`

Generates a human-readable string representation of the normalizer.

Parameters:

Name	Type	Description	Default
`return_output`	`bool`	If true, returns the output text instead of printing it.	`False`

Returns:

Type	Description
`Optional[str]`	A string containing the human-readable representation of the normalizer.

dgf.transform.IdentityNormalizer

Bases: AbstractFeatureNormalizer

A normalizer that simply pass a feature without changing it.

dgf.transform.NumNodesPredicate

Predicate for filtering by number of nodes.

Note the num_nodes field on an InMemoryNodeSet is optional. It is up to the caller to check that it is set.

Returns True if the number of nodes for the nodset_name is in the range [lower, upper): lower <= number of nodes < upper.

Attributes:

Name	Type	Description
`lower`	`int`	The lower bound for the number of nodes. Defaults to np.inf
`upper`	`int`	The upper bound for the number of nodes. Defaults to -np.inf
`nodeset_name`	`Optional[str]`	Optional name of node set to consider. If None (default), will count (sum) all the nodes in all nodesets.

dgf.transform.SoftQuantileNormalizer

Bases: AbstractFeatureNormalizer

Normalizes a numerical feature by replacing it with its soft quantile -0.5.

Soft quantile are different from regular quantiles are follow

When falling in between two quantiles, the value is linearly interpolated.
When falling outside of the quantile ranges (e.g., smaller that the first quantile, or larget that the last quantile), the value is extrapolated.

This normalizer is suitable for numerical features where the quantiles are available in the statistics. The output is a float between -0.5-eps and 0.5+eps, where eps is the extrapolation (generally ~ 1/num quantiles).

Note

For timeseries features, this normalizer will perform some amount of lookahead, i.e. the normalization depends on future feature statistics and leaks some of that information. This can lead to worse-than-expected performance on the validation and test set.

dgf.transform.apply_feature

Applies feature processors to the node and edge sets of a graph.

This function does not check the validity of the output e.g. the returned schema matches the returned value. For more safety, call dgf.validate.validate_graph on the output.

The FeatureSetProcessor should not modify their input values i.e. they have to be functionnal.

Usage Example:

# Create a simple FeatureSetProcessor for the "n1" nodeset.
def process_n1(
      values: Dict[str, np.ndarray],
      schemas: Dict[str, ydf.data.FeatureSchema],
      num_nodes: int) -> Tuple[
          Dict[str, np.ndarray],
          Dict[str, ydf.data.FeatureSchema]
          ]:

  new_values = {
      "f2" : values["f1"] * 3,
      "f3" : np.cast(values["f1"], np.int)
  }
  new_schemas = {
    "f2": schemas["f1"], # "f2" has the same schema as "f1"
    "f3": dgf.data.FeatureSchema(format=dgf.data.FeatureFormat.INTEGER_64)
  }
  return new_values, new_schemas

# Read a graph.
graph, schema = ydf.io.read_graph(...)

# Process the features of the "n1" nodesets and remove any other nodesets.
Don't modify the features of the edgesets.
new_graph, new_schema = ydf.transform.apply_feature(
    graph, schema, {"n1":process_n1})

# Validate the result (make sure your processor works).
ydf.validate.validate_graph(new_graph, new_schema)

Parameters:

Name	Type	Description	Default
`graph`	`Graph`	The input `InMemoryGraph`.	required
`schema`	`GraphSchema`	The `GraphSchema` corresponding to the input graph.	required
`process_nodesets`	`Optional[Dict[str, FeatureSetProcessor]]`	An optional dictionary mapping node set names to `FeatureSetProcessor` instances. Non specified nodesets are removed.	`None`
`process_edgesets`	`Optional[Dict[str, FeatureSetProcessor]]`	An optional dictionary mapping edge set names to `FeatureSetProcessor` instances. Non specified edgeset are removed.	`None`

Returns:

Type	Description
`Tuple[Graph, GraphSchema]`	A tuple containing: - A new `ydf.data.InMemoryGraph` with the transformed features. - A new `ydf.data.GraphSchema` reflecting the schema of the transformed features.

dgf.transform.auto_normalize

Create a generally good GraphNormalizer from feature statistics.

Usage example:

# Schema of the graph
input_schema = dgf.data.GraphSchema(...)
# Statistics about the graph features
input_stats = dgf.data.GraphFeatureStatistics(...)
# Actual graph input data
input_graph = dgf.data.InMemoryGraph(...)

# Instantiate normalizer
normalizer = auto_normalize(input_schema, input_stats)

# Normalize graph features
output_graph = normalizer.normalize_numpy(input_graph)

# Get the schema of the normalized graph
output_schema = normalizer.output_schema()

Applies a set of generally useful transformations to numeric and categorical features based on provided statistics. For example, numeric features can be z-score normalized, and categorical features can be converted to embeddings.

If an input feature cannot be consumed (e.g., not yet implemented) a warning is printed.

This methods instantiates and calls AbstractFeatureNormalizers on each features according to the input semantic+format+stats. For more control, you can manually instantiate and call the AbstractFeatureNormalizers.

Parameters:

Name	Type	Description	Default
`schema`	`GraphSchema`	Schema of the graph features.	required
`stats`	`GraphFeatureStatistics`	Precomputed statistics of the graph features.	required
`config`	`AutoNormalizeConfig`	Configuration for the automatic normalization process.	`AutoNormalizeConfig()`

Returns:

Type	Description
`GraphNormalizer`	A GraphNormalizer instance.

dgf.transform.batch_indices_generator

Generates batches of indices.

If items is an integer, generates batches of indices from 0 to num_items - 1. If items is a numpy array, generates batches from items.

These batches are typically used to feed node indices to a graph sampler.

Usage example:

# Typical batch generation for training
for batch in batch_indices_generator(
      num_nodes=10,
      batch_size=3,
      drop_remainder=True,
      shuffle=True):
  print(batch)
# Example output (order varies due to shuffle):
# [2 8 0]
# [9 1 5]
# [4 6 7]

# Typical batch generation for evaluation
for batch in batch_indices_generator(
      num_nodes=10,
      batch_size=3,
      drop_remainder=False,
      shuffle=False):
  print(batch)
# [0 1 2]
# [3 4 5]
# [6 7 8]
# [9]

Parameters:

Name	Type	Description	Default
`items`	`Union[int, ndarray]`	If an integer, the number of items to sample batches from. The returned indices will be in the range `[0, items)`. If a numpy array, the array to sample batches from.	required
`batch_size`	`int`	The desired size of each batch. If `drop_remainder` is `False`, the last batch may contain fewer than `batch_size` elements.	required
`drop_remainder`	`bool`	If `True`, the last batch will be dropped if it contains fewer than `batch_size` elements. If `False`, all indices are returned, potentially with a smaller last batch.	required
`shuffle`	`bool`	If `True`, the indices are shuffled before batching. If `False`, indices are returned in ascending order (e.g., `[0, ..., j]`, `[j+1, ...]`).	required

Yields:

Type	Description
`ndarray`	A numpy array containing a batch of indices.

dgf.transform.drop_edge_features

Drops all edge features from a graph and its schema.

dgf.transform.drop_edge_features_from_schema

Drops all edge features from a schema.

dgf.transform.filter_graph

Creates an in-memory graph with a subset of nodesets/edgesets/features.

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The input InMemoryGraph.	required
`schema`	`GraphSchema`	The schema defining the subset of nodesets, edgesets, and features.	required

Returns:

Type	Description
`InMemoryGraph`	A new InMemoryGraph containing only the specified nodesets,
`InMemoryGraph`	edgesets, and features.

dgf.transform.filter_graphs

Filters a sequence of graphs based on user defined predicates.

Simple single-thread alternative to built in filter for chaining multiple predicates over a single pass of len(graphs).

Parameters:

Name	Type	Description	Default
`graphs`	`Sequence[InMemoryGraph]`	A sequence of dgf.data.InMemoryGraph objects	required
`predicates`	`Sequence[InMemoryGraphPredicate]`	A sequence of callables that returns a boolean for each input graph.	required
`verbose`	`bool`	Print filtering progress bar.	`True`
`tqdm_desc`	`str`	String passed as description to tqdm progress bar.	`'Filtering Graphs'`

Returns:

Type	Description
`List[List[InMemoryGraph]]`	A list of len(prediates) lists which contain all graphs that satisfy each
`List[List[InMemoryGraph]]`	predicate.

dgf.transform.filter_schema

Extracts a subset of the nodesets/edgesets/features from a schema.

Dangling edgesets (those with source or target nodes removed) are automatically excluded.

Parameters:

Name	Type	Description	Default
`schema`	`GraphSchema`	The original GraphSchema to filter.	required
`filter`	`GraphSchemaFilter`	A GraphSchemaFilter specifying which parts of the schema to keep.	required

Returns:

Type	Description
`GraphSchema`	A new GraphSchema containing only the filtered elements.

dgf.transform.homogeneous_graph_piece_to_nx

Convert InMemoryGraph to an nx.Graph object.

This function is named homogeneous_ because we assume the domain (source) and range (target) of edgeset_name is from/onto nodeset_name. We do not check the schema if this is true.

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The dgf.data.InMemoryGraph to convert.	required
`nodeset_name`	`str`	The name of the nodeset to convert.	required
`edgeset_name`	`str`	The name of the edgeset to convert. We assume domain and range of `edgeset_name` is from/onto `nodeset_name`.	required
`id_feature_name`	`Optional[str]`	The name of the feature to use as the node ID. We use indices on [0, num_nodes - 1] if not provide	`None`
`features_to_keep`	`str \| Sequence[str]`	A string or list of string of features to add to nx node data	`()`
`verbose`		Print conversion progress.	`True`

Returns:

Type	Description
	A networkx Graph object.

dgf.transform.homogenize

Homogenizes a heterogeneous graph into a homogeneous one.

All nodesets on the input graph must have the same feature schemas. Similarly, the all the edgesets of the input graph must have the same feature schemas. If this is not the case (e.g., different nodesets have different features), this alignements can be done with "ydf.transform.apply_feature`.

This function can be applied both on InMemoryGraph (NumPy-based) and JaxInMemoryGraph (JAX-based). This method is jittable.

Usage example:

# Read a graph.
graph, schema = ydf.io.read_graph(...)
new_graph, new_schema = ydf.transform.homogenize(graph, schema)

Parameters:

Name	Type	Description	Default
`graph`	`Graph`	The input `InMemoryGraph`.	required
`schema`	`GraphSchema`	The `GraphSchema` corresponding to the input graph.	required
`homogenized_nodeset_name`	`str`	The name of the new homogenized node set.	`'nodes'`
`homogenized_edgeset_name`	`str`	The name of the new homogenized edge set.	`'edges'`

Returns:

Type	Description
`Tuple[Graph, GraphSchema, Dict[str, int]]`	A tuple containing: - A new `InMemoryGraph` representing the homogenized graph. - A new `GraphSchema` for the homogenized graph. - A mapping from original nodeset names to the starting index of their nodes within the homogenized graph.

dgf.transform.merge_graphs

Merges multiple InMemoryGraph instances into a single graph.

Graphs are concatenated sequentially. Node and edge indices from each graph are offset to prevent collisions in the merged graph.

Padding nodes and edges are added according to the padding configuration. Padding edges are defined to connect to the last node index of their respective destination node set. Therefore, ensure that each node set's padding size is at least one greater than the maximum number of nodes expected before padding.

If padding is None, the edges/nodes are not padded.

If padding is provided, but not large enough to encode all the graphs (e.g., the padding has space for 100 nodes but 120 nodes are provided), returns a InsufficientPaddingError exception.

Parameters:

Name	Type	Description	Default
`graphs`	`List[InMemoryGraph]`	A list of graphs to merge.	required
`schema`	`GraphSchema`	Graph schema.	required
`padding`	`Optional[Padding]`	The padding configuration to apply.	required
`sentinel_offset`	`bool`	Where to include the offset of sentinel nodes / edges added for the padding. Only used when `padding` is set.	`True`

Returns:

Type Description

Tuple[InMemoryGraph, Dict[str, ndarray]]

A tuple containing: - The merged InMemoryGraph. - A dictionary mapping each node set name to a NumPy array. The array at node_set_offsets[node_set_name][i] contains the starting node index of the i-th graph from the input graphs list within the merged graph's node set named node_set_name. If sentinel_offset=True, assuming the graphs contains n values, all the node_set_offsets[node_set_name] will contain n+1 values. This extra value represent the fake node used to pad edges. If sentinel_offset=False, the node_set_offsets[node_set_name] will contain n values.

dgf.transform.propagate_timestamp_to_edges

Propagates timestamps from nodes to edges.

Computes a new edge feature for edgesets that don't have a timestamp, as the maximum value of the timestamps of the two connected nodes.

Usage example:

new_graph, new_schema = dgf.transform.propagate_timestamp_to_edges(
    graph=graph,
    schema=schema,
    target_edgesets=["e1"],
    target_feature="ts",
)

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The input in-memory graph.	required
`schema`	`GraphSchema`	The graph schema.	required
`target_edgesets`	`Optional[List[str]]`	Edgesets to populate with a timestamp. If None, process all edgesets.	`None`
`target_feature`	`str`	The name of the new edge feature. Defaults to "timestamps".	`'timestamps'`

Returns:

Type	Description
`tuple[InMemoryGraph, GraphSchema]`	A tuple containing the new graph and new schema.

dgf.transform.remove_padding_sentinels

Removes the sentinel nodes and edges added by merge_graphs.

The graph and offsets arguments are the two return values from merge_graphs. This method requires merge_graphs to have been called with sentinel_offset=True.

Usage example:

  padded_graph, offsets = merge_graphs([graph], schema, padding,
  sentinel_offset=True)
  unpadded_graph = remove_padding_sentinels(merged_graph, schema, offsets)
  assert unpadded_graph == graph

This method can be used to more easily inspect or plot padded graphs, as padding makes the interpretation harder.

Parameters:

Name	Type	Description	Default
`graph`	`InMemoryGraph`	The padded graph.	required
`schema`	`GraphSchema`	Graph schema.	required
`offsets`	`Dict[str, ndarray]`	Node set offsets returned by `merge_graphs`.	required

Returns:

Type	Description
`InMemoryGraph`	A new `InMemoryGraph` with sentinel nodes and edges removed.

dgf.transform.table2graph

Converts a table (dict of arrays or DataFrame) into an InMemoryGraph and Schema.

The resulting graph will have no edgeset and a single nodeset.

Usage example:

  import numpy as np
  table = {
      "feature_a": np.array([1, 2, 3]),
      "feature_b": np.array([4, 5, 6]),
  }
  graph, schema = dgf.transform.table2graph(table, nodeset_name="my_nodes")

Parameters:

Name	Type	Description	Default
`table`	`Union[Dict[str, ndarray], DataFrame]`	A dictionary of numpy arrays or a pandas DataFrame.	required
`nodeset_name`	`str`	The name of the single nodeset in the output graph.	`'nodes'`
`detect_semantic`	`bool`	Whether to automatically infer the semantic of the features.	`True`

Returns:

Type	Description
`Tuple[InMemoryGraph, GraphSchema]`	A tuple of (InMemoryGraph, GraphSchema).

Raises:

Type	Description
`TypeError`	If the input table is not a dict or pandas DataFrame.
`ValueError`	If the input table is empty, or if the arrays/columns have different lengths.