swh.graph.libs.luigi.topology module

Luigi tasks to analyze, and produce datasets related to, graph topology

This module contains Luigi tasks driving the computation of a topological order, and count the number of paths to every node.

File layout

This assumes a local compressed graph (from swh.graph.luigi.compressed_graph) is present, and generates/manipulates the following files:

base_dir/
    <date>[_<flavor>]/
        topology/
            topological_order_dfs.csv.zst

swh.graph.libs.luigi.topology.inverse_direction(direction: str) → str

class swh.graph.libs.luigi.topology.TopoSort(*args, **kwargs)

Bases: Task

Creates a file that contains all SWHIDs in topological order from a compressed graph.

local_graph_path = <luigi.parameter.PathParameter object>

topology_dir = <luigi.parameter.PathParameter object>

graph_name = <luigi.parameter.Parameter object>

object_types = <luigi.parameter.Parameter object>

direction = <luigi.parameter.ChoiceParameter object>

algorithm = <luigi.parameter.ChoiceParameter object>

requires() → List[Task]

Returns an instance of LocalGraph.

output() → LocalTarget

.csv.zst file that contains the topological order.

run() → None

Runs ‘toposort’ command from tools/topology and compresses

class swh.graph.libs.luigi.topology.ComputeGenerations(*args, **kwargs)

Bases: Task

Creates a file that contains all SWHIDs in topological order from a compressed graph.

local_graph_path = <luigi.parameter.PathParameter object>

topology_dir = <luigi.parameter.PathParameter object>

graph_name = <luigi.parameter.Parameter object>

object_types = <luigi.parameter.Parameter object>

direction = <luigi.parameter.ChoiceParameter object>

requires() → List[Task]

Returns an instance of LocalGraph.

output() → dict[str, Target]

.csv.zst file that contains the topological order.

run() → None

Runs ‘toposort’ command from tools/topology and compresses

class swh.graph.libs.luigi.topology.UploadGenerationsToS3(*args, **kwargs)

Bases: Task

Uploads the output of ComputeGenerations to S3

local_graph_path = <luigi.parameter.PathParameter object>

topology_dir = <luigi.parameter.PathParameter object>

dataset_name = <luigi.parameter.Parameter object>

graph_name = <luigi.parameter.Parameter object>

object_types = <luigi.parameter.Parameter object>

direction = <luigi.parameter.ChoiceParameter object>

requires() → Task

Returns an instance of ComputeGenerations.

output() → List[Target]

Returns .bitstream and .bin paths on S3.

run() → None

Copies the files

class swh.graph.libs.luigi.topology.CountPaths(*args, **kwargs)

Bases: Task

Creates a file that lists:

• the number of paths leading to each node, and starting from all leaves, and

• the number of paths leading to each node, and starting from all other nodes

Counts include the trivial self-path (zero-length path from a node to itself).

local_graph_path = <luigi.parameter.PathParameter object>

topology_dir = <luigi.parameter.PathParameter object>

graph_name = <luigi.parameter.Parameter object>

write_parquet = <luigi.parameter.BoolParameter object>

object_types = <luigi.parameter.Parameter object>

direction = <luigi.parameter.ChoiceParameter object>

property resources

Return the estimated RAM use of this task.

requires() → Dict[str, Task]

Returns an instance of LocalGraph and one of ComputeGenerations.

output() → dict[str, LocalTarget]

Parquet directory (if enabled) and binary arrays with path counts.

run() → None

Runs ‘count_paths’ command from tools/topology

class swh.graph.libs.luigi.topology.CountDescendants(*args, **kwargs)

Bases: Task

Counts the number of unique descendants (or ancestors) of each node using a probabilistic HyperLogLog counter.

With --direction forward, for each node counts unique nodes reachable from it. With --direction backward, counts unique nodes that can reach it.

local_graph_path = <luigi.parameter.PathParameter object>

topology_dir = <luigi.parameter.PathParameter object>

graph_name = <luigi.parameter.Parameter object>

object_types = <luigi.parameter.Parameter object>

write_parquet = <luigi.parameter.BoolParameter object>

direction = <luigi.parameter.ChoiceParameter object>

rsd = <luigi.parameter.FloatParameter object>

seed = <luigi.parameter.OptionalIntParameter object>

property resources

Return the estimated RAM use of this task.

requires() → Dict[str, Task]

Returns an instance of LocalGraph and one of ComputeGenerations.

output() → dict[str, LocalTarget]

Parquet directory (if enabled) and binary array with descendant counts.

run() → None

Runs ‘count_descendants’ command from tools/topology

class swh.graph.libs.luigi.topology.PathCountsParquetToS3(*args, **kwargs)

Bases: _ParquetToS3ToAthenaTask

Reads the CSV from CountPaths, converts it to ORC, upload the ORC to S3, and create an Athena table for it.

topology_dir = <luigi.parameter.PathParameter object>

object_types = <luigi.parameter.Parameter object>

direction = <luigi.parameter.ChoiceParameter object>

dataset_name = <luigi.parameter.Parameter object>

s3_athena_output_location = <swh.export.luigi.S3PathParameter object>

requires() → CountPaths

Returns corresponding CountPaths instance