swh.graph.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

class swh.graph.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>

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

topological_order_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.luigi.topology.UploadGenerationsToS3(*args, **kwargs)

Bases: Task

Uploads the output of ComputeGenerations to S3

local_graph_path = <luigi.parameter.PathParameter object>

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

Singleton paths are not counted.

local_graph_path = <luigi.parameter.PathParameter object>

topological_order_dir = <luigi.parameter.PathParameter object>

graph_name = <luigi.parameter.Parameter 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 TopoSort.

output() → LocalTarget

.csv.zst file that contains the counts.

nb_lines()

run() → None

Runs ‘count_paths’ command from tools/topology and compresses

class swh.graph.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.

topological_order_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