swh.graph.luigi.compressed_graph module#

Luigi tasks for compression#

This module contains Luigi tasks, as an alternative to the CLI that can be composed with other tasks, such as swh-dataset’s.

It implements the task DAG described in Compression steps.

Unlike the CLI, this requires the graph to be named graph.

Filtering#

The object_types parameter (--object-types on the CLI) specifies the set of node types to read from the dataset export, and it defaults to all types: ori,snp,rel,rev,dir,cnt.

Because the dataset export is keyed by edge sources, some objects without any of these types will be present in the input dataset. For example, if exporting ori,snp,rel,rev, root Directory of every release and revision will be present, though without its labels (as well as a few Content objects pointed by some Releases).

File layout#

In addition to files documented in Graph compression (eg. graph.graph, graph.mph, …), tasks in this module produce this directory structure:

base_dir/
    <date>[_<flavor>]/
        compressed/
            graph.graph
            graph.mph
            ...
            meta/
                export.json
                compression.json

graph.meta/export.json is copied from the ORC dataset exported by swh.dataset.luigi.

graph.meta/compression.json contains information about the compression itself, for provenance tracking. For example:

[
    {
        "steps": null,
        "export_start": "2022-11-08T11:00:54.998799+00:00",
        "export_end": "2022-11-08T11:05:53.105519+00:00",
        "object_types": [
            "origin",
            "origin_visit"
        ],
        "hostname": "desktop5",
        "conf": {},
        "tool": {
            "name": "swh.graph",
            "version": "2.2.0"
        },
        "commands": [
            {
                "command": [
                    "bash",
                    "-c",
                    "java it.unimi.dsi.big.webgraph.labelling.BitStreamArcLabelledImmutableGraph --list ..."
                ],
                "cgroup": "/sys/fs/cgroup/user.slice/user-1002.slice/user@1002.service/app.slice/swh.graph@103038/bash@0",
                "cgroup_stats": {
                    "memory.events": "low 0\nhigh 0\nmax 0\noom 0\noom_kill 0\noom_group_kill 0",
                    "memory.events.local": "low 0\nhigh 0\nmax 0\noom 0\noom_kill 0\noom_group_kill 0",
                    "memory.swap.current": "0",
                    "memory.zswap.current": "0",
                    "memory.swap.events": "high 0\nmax 0\nfail 0",
                    "cpu.stat": "usage_usec 531350\nuser_usec 424286\nsystem_usec 107063\n...",
                    "memory.current": "614400",
                    "memory.stat": "anon 0\nfile 110592\nkernel 176128\nkernel_stack 0\n...",
                    "memory.numa_stat": "anon N0=0\nfile N0=110592\nkernel_stack N0=0\n...",
                    "memory.peak": "49258496"
                }
            }
        ]
    }
]

When the compression pipeline is run in separate steps, each of the steps is recorded as an object in the root list.

S3 layout#

As .bin files are meant to be accessed randomly, they are uncompressed on disk. However, this is undesirable on at-rest/long-term storage like on S3, because some are very sparse (eg. graph.property.committer_timestamp.bin can be quickly compressed from 300 to 1GB).

Therefore, these files are compressed to .bin.zst, and need to be decompressed when downloading.

The layout is otherwise the same as the file layout.

class swh.graph.luigi.compressed_graph.ObjectTypesParameter(*args, **kwargs)[source]#

Bases: Parameter

A parameter type whose value is either * or a set of comma-separated object types (eg. ori,snp,rel,rev,dir,cnt).

parse(s: str) → List[str][source]#

Parse an individual value from the input.

The default implementation is the identity function, but subclasses should override this method for specialized parsing.

Parameters:: x (str) – the value to parse.
Returns:: the parsed value.

serialize(value: List[str]) → str[source]#

Opposite of parse().

Converts the value x to a string.

Parameters:: x – the value to serialize.

class swh.graph.luigi.compressed_graph.ExtractNodes(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = -2#

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

INPUT_FILES: Set[str] = {}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.edges.count.txt', '.edges.stats.txt', '.labels.count.txt', '.labels.csv.zst', '.nodes.count.txt', '.nodes.csv.zst', '.nodes.stats.txt'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Mph(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = -1#

INPUT_FILES: Set[str] = {'.nodes.csv.zst'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.mph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.ConvertMph(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 0#

INPUT_FILES: Set[str] = {'.mph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.cmph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Bv(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 1#

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

INPUT_FILES: Set[str] = {'.cmph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-base.graph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.BvOffsets(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 2#

INPUT_FILES: Set[str] = {'-base.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-base.offsets'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.BvEf(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 3#

INPUT_FILES: Set[str] = {'-base.offsets'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-base.ef'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Bfs(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 4#

INPUT_FILES: Set[str] = {'-base.ef', '-base.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-bfs.order'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.PermuteAndSimplifyBfs(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 6#

INPUT_FILES: Set[str] = {'-base.ef', '-base.graph', '-bfs.order'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-bfs-simplified.graph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.BfsOffsets(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 7#

INPUT_FILES: Set[str] = {'-bfs-simplified.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-bfs-simplified.offsets'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Llp(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 8#

INPUT_FILES: Set[str] = {'-bfs-simplified.graph', '-bfs-simplified.offsets'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-llp.order'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.PermuteLlp(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 10#

INPUT_FILES: Set[str] = {'-base.graph', '-base.offsets', '-bfs.order', '-llp.order'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.graph', '.offsets', '.properties'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Obl(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 11#

INPUT_FILES: Set[str] = {'.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.obl'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.ComposeOrders(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 9#

INPUT_FILES: Set[str] = {'-bfs.order', '-llp.order'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.order'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Stats(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 12#

INPUT_FILES: Set[str] = {'.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.stats'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Transpose(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 13#

INPUT_FILES: Set[str] = {'.graph', '.obl'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-transposed.graph', '-transposed.properties'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.TransposeObl(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 14#

INPUT_FILES: Set[str] = {'-transposed.graph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'-transposed.obl'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.Maps(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 15#

INPUT_FILES: Set[str] = {'.mph', '.nodes.csv.zst', '.order'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.node2swhid.bin'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.ExtractPersons(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 16#

INPUT_FILES: Set[str] = {}#: Dependencies of this step.

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

OUTPUT_FILES: Set[str] = {'.persons.csv.zst'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.MphPersons(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 17#

INPUT_FILES: Set[str] = {'.persons.csv.zst'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.persons.mph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.NodeProperties(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 18#

INPUT_FILES: Set[str] = {'.mph', '.node2swhid.bin', '.order', '.persons.mph'}#: Dependencies of this step.

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

OUTPUT_FILES: Set[str] = {'.property.author_id.bin', '.property.author_timestamp.bin', '.property.author_timestamp_offset.bin', '.property.committer_id.bin', '.property.committer_timestamp.bin', '.property.committer_timestamp_offset.bin', '.property.content.is_skipped.bin', '.property.content.length.bin', '.property.message.bin', '.property.message.offset.bin', '.property.tag_name.bin', '.property.tag_name.offset.bin'}#: List of files which this task produces, without the graph name as prefix.

output() → List[LocalTarget][source]#: Returns a list of luigi targets matching OUTPUT_FILES.

class swh.graph.luigi.compressed_graph.MphLabels(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 19#

INPUT_FILES: Set[str] = {'.labels.csv.zst'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.labels.mph'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.FclLabels(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 20#

INPUT_FILES: Set[str] = {'.labels.csv.zst', '.labels.mph'}#: Dependencies of this step.

OUTPUT_FILES: Set[str] = {'.labels.fcl.bytearray', '.labels.fcl.pointers', '.labels.fcl.properties'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.EdgeLabels(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 21#

INPUT_FILES: Set[str] = {'-transposed.properties', '.graph', '.labels.mph', '.mph', '.node2swhid.bin', '.order', '.properties'}#: Dependencies of this step.

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

OUTPUT_FILES: Set[str] = {'-labelled.labeloffsets', '-labelled.labels', '-labelled.properties', '-transposed-labelled.labeloffsets', '-transposed-labelled.labels', '-transposed-labelled.properties'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.EdgeLabelsObl(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 22#

INPUT_FILES: Set[str] = {'-labelled.labeloffsets', '-labelled.labels', '-labelled.properties'}#: Dependencies of this step.

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

OUTPUT_FILES: Set[str] = {'-labelled.labelobl'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.EdgeLabelsTransposeObl(*args, **kwargs)[source]#

Bases: _CompressionStepTask

STEP: CompressionStep = 23#

INPUT_FILES: Set[str] = {'-transposed-labelled.labeloffsets', '-transposed-labelled.labels', '-transposed-labelled.properties'}#: Dependencies of this step.

EXPORT_AS_INPUT: bool = True#: Whether this task should depend directly on LocalExport. If not, it is assumed it depends transitiviely via one of the tasks returned by requires().

OUTPUT_FILES: Set[str] = {'-transposed-labelled.labelobl'}#: List of files which this task produces, without the graph name as prefix.

class swh.graph.luigi.compressed_graph.CompressGraph(*args, **kwargs)[source]#

Bases: Task

local_export_path = <luigi.parameter.PathParameter object>#

graph_name = <luigi.parameter.Parameter object>#

local_graph_path = <luigi.parameter.PathParameter object>#

batch_size = <luigi.parameter.IntParameter object>#

rust_executable_dir = <luigi.parameter.Parameter object>#

object_types = <swh.graph.luigi.compressed_graph.ObjectTypesParameter object>#

requires() → List[Task][source]#: Returns a LocalExport task, and leaves of the compression dependency graph

output() → List[LocalTarget][source]#: Returns the meta/*.json targets

run()[source]#

Runs the full compression pipeline, then writes meta/compression.json

This does not support running individual steps yet.

class swh.graph.luigi.compressed_graph.UploadGraphToS3(*args, **kwargs)[source]#

Bases: Task

Uploads a local compressed graphto S3; creating automatically if it does not exist.

Example invocation:

luigi --local-scheduler --module swh.graph.luigi UploadGraphToS3                 --local-graph-path=graph/                 --s3-graph-path=s3://softwareheritage/graph/swh_2022-11-08/compressed/

local_graph_path = <luigi.parameter.PathParameter object>#

s3_graph_path = <swh.dataset.luigi.S3PathParameter object>#

parallelism = <luigi.parameter.IntParameter object>#

requires() → List[Task][source]#: Returns a CompressGraph task that writes local files at the expected location.

output() → List[Target][source]#: Returns stamp and meta paths on S3.

run() → None[source]#: Copies all files: first the graph itself, then meta/compression.json.

class swh.graph.luigi.compressed_graph.DownloadGraphFromS3(*args, **kwargs)[source]#

Bases: Task

Downloads a local dataset graph from S3.

This performs the inverse operation of UploadGraphToS3

Example invocation:

luigi --local-scheduler --module swh.graph.luigi DownloadGraphFromS3                 --local-graph-path=graph/                 --s3-graph-path=s3://softwareheritage/graph/swh_2022-11-08/compressed/

local_graph_path = <luigi.parameter.PathParameter object>#

s3_graph_path = <swh.dataset.luigi.S3PathParameter object>#

requires() → List[Task][source]#: Returns a UploadGraphToS3 task that writes local files to S3.

output() → List[Target][source]#: Returns stamp and meta paths on the local filesystem.

run() → None[source]#: Copies all files: first the graph itself, then meta/compression.json.

class swh.graph.luigi.compressed_graph.LocalGraph(*args, **kwargs)[source]#

Bases: Task

Task that depends on a local dataset being present – either directly from ExportGraph or via DownloadGraphFromS3.

local_graph_path = <luigi.parameter.PathParameter object>#

compression_task_type = <luigi.parameter.TaskParameter object>#

requires() → List[Task][source]#: Returns an instance of either CompressGraph or DownloadGraphFromS3 depending on the value of compression_task_type.

output() → List[Target][source]#: Returns stamp and meta paths on the local filesystem.