Distributed OWL Reasoners

[Demirrr]

Description

1. At each node, we store the fragment of ABOX and the full TBOX/RBOX.
2. At each node, we load the data into a reasoner.
3. Send a query to nodes, e.g. Female
4. Return a set of instance per node
5. Aggregate the results

Here is an example over csv data.

Step 1: Create Data Fragments

python -c "import pandas as pd; \
pd.DataFrame([{'id': 1, 'name': 'Alice', 'country': 'Germany'}, {'id': 2, 'name': 'Bob', 'country': 'France'}]).to_csv('db_part_1.csv', index=False); \
pd.DataFrame([{'id': 3, 'name': 'Charlie', 'country': 'Germany'}, {'id': 4, 'name': 'David', 'country': 'Spain'}]).to_csv('db_part_2.csv', index=False)"

---

Step 2: Start the Infrastructure

Open three terminals and run these commands. The --resources flag tells Ray which terminal has which file.

Terminal 1 (Head Node + Fragment 1)

ray start --head --port=6379 --num-cpus=1 --resources='{"server_alpha": 1}'

Terminal 2 (Worker Node + Fragment 2)

ray start --address='192.168.2.225:6379' --num-cpus=1 --resources='{"server_beta": 1}'

Terminal 3 (Client)

Keep this terminal open to run your Python code.

import ray
import pandas as pd

ray.init(address='auto')

@ray.remote
class DatabaseShard:
    def __init__(self, shard_id, file_path):
        self.shard_id = shard_id
        # Load data into RAM once
        self.df = pd.read_csv(file_path)
        print(f"--- Shard {shard_id} initialized and data loaded into RAM ---")

    def query(self, country):
        print(f"Shard {self.shard_id} searching for: {country}")
        result = self.df[self.df['country'] == country]
        return result.to_dict('records')

if __name__ == "__main__":
    # 1. Instantiate Actors on specific nodes based on resources
    shard_1 = DatabaseShard.options(resources={"server_alpha": 1}).remote("Alpha", "db_part_1.csv")
    shard_2 = DatabaseShard.options(resources={"server_beta": 1}).remote("Beta", "db_part_2.csv")

    # 2. Perform a distributed search
    search_target = "Germany"
    print(f"\nRequesting '{search_target}' from all shards...")

    # Both shards search their own memory in parallel
    results = ray.get([
        shard_1.query.remote(search_target),
        shard_2.query.remote(search_target)
    ])

    # 3. Merge results (The 'Reduce' step)
    flat_results = [item for sublist in results for item in sublist]
    print(f"\nFound {len(flat_results)} records:")
    for record in flat_results:
        print(f" - {record['name']} lives in {record['country']}")

[Demirrr]

A version working https://github.com/dice-group/owlapy/blob/ddp_owl_reasoning/ddp_reasoning.py

[Demirrr]

After considertation, I have realised that due to storing fragmented data (e.g. same Tbox and a split of Abox) on each process/node would lead a reasoner to generate incomplete results if the query is not one-hop query.
For instance, answering exist r. C requires two steps, C(x) and y r.x. Yet, C(x) and y r.x must be in the same fragment for this idea to generate complete answer.

We need to think about divide the graph more intelligenty if possible into different processes to make distributed reasoners to work.

A one way to address this issue would be OWL Class Expression Decomposition in a fashion akin to
Neural Reasoning for Robust Instance Retrieval in SHOIQ(https://dl.acm.org/doi/full/10.1145/3731443.3771348)

I will continue on this idea in this issue by regularly updating this comment :)

[Demirrr]

@LckyLke random seed doesnt seem to work does it ?

[LckyLke]

@LckyLke random seed doesnt seem to work does it ?

I will try it

[LckyLke]

@LckyLke random seed doesnt seem to work does it ?

I guess this is due to the set order not being controlled by the normal seed?
We can sort this concepts list before shuffling (since this shuffle will be controlled by the seed) which should fix it:

    concepts = list(
        chain(
            nc,                           # named concepts (C)
            nnc,                          # negated named concepts (¬C)
            unions_nc_star,               # NC* UNION NC*
            intersections_nc_star,        # NC* INTERSECTION NC*
            exist_nc_star,                # ∃ r.C
            for_all_nc_star,              # ∀ r.C
            min_cardinality_nc_star_1, min_cardinality_nc_star_2, min_cardinality_nc_star_3,
            max_cardinality_nc_star_1, max_cardinality_nc_star_2, max_cardinality_nc_star_3,
            exist_nominals
        )
    )

I hope this fixes it: 6827a8e

[LckyLke]

I think we also have to set PYTHONHASHSEED to ensure reproducibility:

e.g.

if __name__ == "__main__":
    args = get_default_arguments()
    
    # Ensure deterministic hash/set behavior by setting PYTHONHASHSEED
    if os.environ.get("PYTHONHASHSEED") != str(args.seed):
        os.environ["PYTHONHASHSEED"] = str(args.seed)
        sys.stdout.flush()
        os.execv(sys.executable, [sys.executable] + sys.argv)
        
    execute(args)

[Demirrr]

Thank you!

[LckyLke]

I will update the script than 👍🏼