Streaming parsers

[gromgull]

Currently all parsers write into a graph. For cases where you want to stream-process RDF, it would be nice if triples could be handled as they come in.

Two fold:

1 Most of our parsers have a Triple Sink object:

• https://github.com/RDFLib/rdflib/blob/master/rdflib/plugins/parsers/nt.py#L7
• https://github.com/RDFLib/rdflib/blob/master/rdflib/plugins/parsers/ntriples.py#L41
• https://github.com/RDFLib/rdflib/blob/master/rdflib/plugins/parsers/notation3.py#L1572

Define this interface centrally and unify

2 Make sure we read the input stream as a stream, and do not read the whole thing into a string :)

[uholzer]

The same would be useful for serializers. This would one allow to easily impelement convertors between different serializations. Of course, this would not easily work for serializers that want to have the triples in a certain order.

[rlk-ama]

Hi,
have you done anything in that direction yet ?

[KMax]

It is quite hard to build a reliable data processing pipeline with rdflib due to lack of the streaming support, i.e., everything have to be loaded into the memory first, therefore you cannot predict how much memory you need in advance. See my discussion in https://github.com/orgs/Jelly-RDF/discussions/97 where I was exploring Jelly format to tackle the issue.

One example use case would be to process an N-Triples file to fetch which ontology elements are used in there, so that this information could be used later for some query optimization or search purposes.

[Ostrzyciel]

In pyjelly we do have a streaming parser for the Jelly format that works with RDFLib. It can output to an RDFLib sink (graph or dataset), but we also have an extra method that returns a generator of RDF statements: https://github.com/Jelly-RDF/pyjelly/blob/e60e52efa3c47906517b361511dfe309a26171b0/pyjelly/integrations/rdflib/parse.py#L450

I guess you could hack together something like this with the current RDFLib parser, if you passed in a sink that would ingest the statements and make them immediately available as a generator. Or at least send these statements to some callback function. In Jelly we essentially circumvented that entirely.

[nichtich]

How about something like this?

from rdflib.graph import Graph

class TripleStream(Graph):
    def __init__(self, triple_callback):
        super().__init__()
        self.triple_callback = triple_callback
    def add(self, triple):
        self.triple_callback(triple)

g = TripleStream(lambda t: print(t))
g.parse(source="./example.ttl")

[nichtich]

Extended version with iterator instead of callback:

from queue import Queue
import threading
from rdflib.graph import Graph

def parse_rdf_triples(source, buffer=10):
    q = Queue(maxsize=buffer)

    class TripleStream(Graph):
        def __init__(self):
            super().__init__()
        def add(self, triple):
            q.put(triple)

    def task():
        g = TripleStream()
        try:
            g.parse(source=source)
        except Exception as e:
            q.put(e)
        q.put(False)

    t = threading.Thread(target=task)
    t.start()

    while True:
        triple = q.get(timeout=5)
        if not triple:
            break
        if type(triple) is Exception:
            raise Exception
        yield triple

    t.join()

for triple in parse_rdf_triples(source="./example.ttl"):
    print(triple)