Custom Connectors

The engine talks to two interfaces. Implement either to add a system.

A sink

Four methods. delivery tells the engine whether to drive resume from the sink (exactly-once) or let the source resume itself (at-least-once).

from pyrula.pipelines import DeliveryGuarantee, Sink

class StdoutSink(Sink):
    delivery = DeliveryGuarantee.AT_LEAST_ONCE

    def open(self, schema): ...
    def write_batch(self, batch, checkpoint):
        import pyarrow as pa
        print(pa.table(batch))
    def resume_checkpoint(self):
        return None
    def close(self): ...

For exactly-once, write_batch must commit the data and the checkpoint together; raising means nothing committed. resume_checkpoint returns the last checkpoint you committed, or None for a fresh start.

A source

from pyrula.pipelines import Source

class CounterSource(Source):
    def open(self):
        self._n = 0
    def poll(self):
        if self._n >= 100:
            return None
        import pyarrow as pa
        batch = pa.table({"n": [self._n]})
        self._n += 1
        return batch, {"counter": {"n": self._n}}
    def seek(self, checkpoint):
        self._n = checkpoint["counter"]["n"]
    def commit(self, checkpoint): ...
    def close(self): ...

poll returns an Arrow batch and a checkpoint, or None when there is nothing right now. The checkpoint is yours to shape; keep it JSON-serializable. seek positions the source so the record after the checkpoint is next.

Errors

A sink write that keeps failing is retried with backoff up to max_retries, then the engine raises PipelineError with the last committed checkpoint attached, which is where you resume from.