pw.io.pulsar

This module is available when using one of the following licenses only: Pathway Scale, Pathway Enterprise.

Authentication

The connectors support the two most common Pulsar authentication mechanisms. The mechanism is passed in the auth parameter of both read and write; None (the default) connects without authentication:

• Token authentication: auth=TokenAuthentication(token) presents the token (for example, a JWT) to the broker.
• OAuth2 client credentials: auth=OAuth2Authentication(issuer_url, credentials_url, audience=..., scope=...) obtains an access token from the provider and refreshes it automatically.

Whether the connection is encrypted is determined by the scheme of the service URI passed as uri (the pulsar+ssl scheme enables TLS). For encrypted connections, the CA certificate used to verify the broker can be given in tls_settings (TLSSettings.root_cert_path); TLSSettings.trust_certificates disables the verification for development setups. Mutual TLS (client certificate) authentication is not supported.

Delivery semantics

The input connector has two reading mechanisms, selected by the subscription_type parameter of read.

The partition-reader mode (subscription_type="reader") treats every partition of the topic as an independent replayable log, the way Kafka consumers treat theirs: the partitions are split between the workers, each is consumed from an explicit position, and the per-partition positions of the delivered messages are recorded in the persistence checkpoint. A restarted pipeline resumes every partition exactly after the last checkpointed position, re-reading the uncheckpointed tail from the topic itself — so the recovery neither loses messages nor processes them twice, regardless of how many workers the pipeline runs or where the crash landed. This is the only mechanism allowed with persistence (and the implicit choice when persistence is enabled), because it is the only one whose recovery does not depend on broker-side state. The operational requirement it inherits from the Kafka model: the topic retention must be long enough to still hold the uncheckpointed tail when the pipeline restarts.

The subscription modes ("shared", "key_shared", "exclusive", "failover") read through a broker-side subscription whose cursor advances as the messages are consumed. They are the right choice for non-persistent streaming — "shared" in particular scales beyond the partition count — but a subscription cursor cannot replay a precise message range, so these modes are rejected when persistence is enabled.

The output connector attaches the Pathway minibatch time and the type of each update as the pathway_time and pathway_diff message properties, and assigns every message a partition key, so the updates of a single row stay in one partition and keep their order. The write delivery is at-least-once: a retried send after a broker hiccup may duplicate a message.

Performance

The numbers below come from end-to-end benchmarks: for the write side, Pathway reads a CSV dataset, does basic per-row processing, and publishes every row to a Pulsar topic as one JSON message; for the read side, Pathway drains a prefilled topic into a null sink. They therefore reflect Pathway + the input source + Pulsar together, out of the box, not Pulsar’s standalone ceiling. Every run is verified for correctness: the number of messages committed to (or delivered from) the topic must equal the row count exactly.

Hardware. A single-socket AMD Ryzen 9 5900X (Zen 3, 12 cores / 24 threads, one NUMA node), 125 GiB of RAM, with an NVMe SSD backing the broker storage. Pulsar ran as the stock apachepulsar/pulsar:4.0.4 Docker image in the standalone mode, at its default configuration. The Pathway and Pulsar containers were each pinned to their own core-complex die (6 cores with a private 32 MiB L3).

Write throughput. End-to-end wall-clock time to read a 20 M-row, 64-shard CSV dataset (≈ 0.93 GB) and publish every row to a Pulsar topic, swept over the number of Pathway workers (median of 3 runs):

Pulsar publish throughput by worker count

Publishing to a partitioned topic performs about the same at 1–2 workers and 10–12% better at 4–8 workers, where the parallel per-partition write paths relieve the broker (≈ 694 000 rows/s on 8 workers with 8 partitions).

Read throughput. Time to drain a topic prefilled with the same 20 M messages (median of 3 runs). The two reading mechanisms differ considerably. Through a shared subscription on a non-partitioned topic:

Pulsar read throughput, shared subscription

In the partition-reader mode (subscription_type="reader") on a topic with 8 partitions, both without and with persistence enabled:

Pulsar read throughput, partition-reader mode, 8 partitions

The reader mode is about 4× faster per worker because a shared subscription must acknowledge every delivered message individually to the broker, while the reader mode tracks the read positions on the client side and sends no acknowledgements at all. Persistence comes for free at 1–2 workers and costs 15–20% at 4–8 workers, where the drain is short enough for the fixed snapshotting overhead to show.

For the full methodology, dataset generator, and reproduction steps, see the pulsar-bulk-write and pulsar-bulk-read directories of the Pathway benchmarks repository.

class OAuth2Authentication(issuer_url, credentials_url, audience=None, scope=None)

import pathway as pw
auth = pw.io.pulsar.OAuth2Authentication(
    issuer_url="https://auth.streamnative.cloud/",
    credentials_url="file:///path/to/credentials.json",
    audience="urn:sn:pulsar:my-org:my-instance",
)

class TokenAuthentication(token)

import pathway as pw
auth = pw.io.pulsar.TokenAuthentication("my-jwt-token")

read(uri, topic, *, schema=None, format='raw', mode='streaming', subscription_name=None, subscription_type=None, autocommit_duration_ms=1500, json_field_paths=None, start_from='beginning', start_from_timestamp_ms=None, auth=None, tls_settings=None, name=None, max_backlog_size=None, debug_data=None, **kwargs)

sourceReads data from an Apache Pulsar topic.

The connector has two reading mechanisms, chosen by subscription_type: the Kafka-like partition-reader mode (the only one allowed with persistence, where it recovers without losing or duplicating messages) and the broker-side subscription modes. Their semantics and trade-offs are described in the Delivery semantics section above.

In the static mode the connector reads the messages that exist in the topic at the start of the computation and finishes afterwards, always through the partition-reader mechanism; no cursor is left behind on the broker.

There are three formats supported: "plaintext", "raw", and "json".

For the "raw" format, the payload is read as raw bytes and added directly to the table. In the "plaintext" format, the payload is decoded from UTF-8 and stored as plain text. In both cases, the table will have a "data" column representing the payload.

If "json" is chosen, the connector parses the message payload as JSON and creates table columns based on the schema provided in the schema parameter.

• Parameters
  • uri (str) – The Pulsar service URI, e.g. pulsar://localhost:6650 or pulsar+ssl://my-cluster:6651 for a TLS-encrypted connection.
  • topic (str) – The name of the topic to read from.
  • schema (type[Schema] | None) – Schema of the resulting table. Required for the "json" format.
  • format (Literal['plaintext', 'raw', 'json']) – Format of the incoming messages: "plaintext", "raw", or "json".
  • mode (Literal['streaming', 'static']) – Denotes how the engine polls the topic for the new data. If set to "streaming", it waits for new messages indefinitely. Otherwise, in the "static" mode, it reads the messages that are present in the topic at the start of the computation and finishes.
  • subscription_name (str | None) – The name of the Pulsar subscription to attach to, for the subscription-based reading mechanisms. Providing an explicit name creates a durable subscription: its cursor survives pipeline redeployments and can be inspected by external tools, but it also pins the topic backlog on the broker until the subscription is removed, so its lifecycle is the user’s responsibility. If not set, the connector generates a per-run name and subscribes non-durably, leaving no state behind on the broker; multi-process runs of the "shared" and "key_shared" types require an explicit name, because every process must attach to one shared subscription. The partition-reader mode does not create broker-side subscriptions and ignores this parameter.
  • subscription_type (Optional[Literal['reader', 'shared', 'key_shared', 'exclusive', 'failover']]) – The reading mechanism. "reader" is the Kafka-like partition-reader mode — required for (and implied by) persistence. The subscription modes are "shared" (messages distributed between the workers without ordering guarantees), "key_shared" (distributed by the hash of the message key, so the messages with equal keys are processed in order by one worker), "exclusive" and "failover" (a single active consumer, full order, one reading worker). None (the default) selects "reader" when persistence is enabled and "shared" otherwise. The static mode always uses the partition-reader mechanism and ignores this parameter.
  • autocommit_duration_ms (int | None) – The maximum time between two commits. Every autocommit_duration_ms milliseconds, the updates received by the connector are committed and pushed into Pathway’s computation graph.
  • json_field_paths (dict[str, str] | None) – If the format is "json", this field allows to map field names into path in the read json object. For the field which require such mapping, it should be given in the format <field_name>: <path to be mapped>, where the path to be mapped needs to be a JSON Pointer (RFC 6901).
  • start_from (Literal['beginning', 'end', 'timestamp']) – The position to start reading from, if the subscription does not exist yet: "beginning" reads the topic from the earliest available message, "end" reads only the messages published after the computation start, and "timestamp" delivers only the messages whose publish timestamp is at least start_from_timestamp_ms (the earlier messages are consumed and skipped). If the subscription already exists, the reading continues from its cursor and the earlier messages are not re-read. "end" cannot be combined with persistence — its position would be re-resolved at every restart, losing the downtime window; use "timestamp" with an explicit timestamp instead.
  • start_from_timestamp_ms (int | None) – The publish timestamp, in milliseconds since the UNIX epoch, to start reading from. Requires start_from="timestamp".
  • auth (TokenAuthentication | OAuth2Authentication | None) – The authentication mechanism: TokenAuthentication, OAuth2Authentication, or None for clusters without authentication.
  • tls_settings (TLSSettings | None) – TLS connection settings. Use TLSSettings to provide the CA certificate used to verify the broker (root_cert_path) or, for development setups, to accept any broker certificate (trust_certificates=True). The connection is encrypted when the uri uses the pulsar+ssl:// scheme. Mutual TLS (client certificate) authentication is not supported.
  • name (str | None) – A unique name for the connector. If provided, this name will be used in logs and monitoring dashboards. Additionally, if persistence is enabled, it will be used as the name for the snapshot that stores the connector’s progress.
  • max_backlog_size (int | None) – Limit on the number of entries read from the input source and kept in processing at any moment. Reading pauses when the limit is reached and resumes as processing of some entries completes. Useful with large sources that emit an initial burst of data to avoid memory spikes.
  • debug_data – Static data replacing original one when debug mode is active.
• Returns
  Table – The table read.

Example:

Consider a topic "measurements" on a broker running locally, with JSON messages of the form {"sensor_id": "front-door", "temperature": 21.5}. To parse such messages into a two-column table, describe their fields with a schema and pass it together with the "json" format — the connector then creates one column per schema field:

import pathway as pw
class InputSchema(pw.Schema):
    sensor_id: str
    temperature: float
table = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="json",
    schema=InputSchema,
)

If the payloads are not JSON, the "plaintext" format reads each message into a single data column as a UTF-8 string (and "raw" does the same without decoding, producing bytes). No schema is needed:

table = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="plaintext",
)

By default the reading starts from the earliest available message and continues indefinitely. For a bounded, batch-style computation, the "static" mode reads only the messages present in the topic at the start and then finishes the pipeline:

table = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="json",
    schema=InputSchema,
    mode="static",
)

A token-authenticated read from a TLS-protected cluster, interested only in the messages published after the computation start:

table = pw.io.pulsar.read(
    "pulsar+ssl://my-cluster.example.com:6651",
    "measurements",
    format="json",
    schema=InputSchema,
    auth=pw.io.pulsar.TokenAuthentication("my-jwt-token"),
    start_from="end",
)

To replay the history from a specific moment instead — for example, the last hour — pass the publish timestamp to start from. The messages published earlier are skipped:

import time
table = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="json",
    schema=InputSchema,
    start_from="timestamp",
    start_from_timestamp_ms=int(time.time() * 1000) - 3600 * 1000,
)

The reading position lives in the broker-side subscription. By default the connector generates a subscription name; providing an explicit one makes the position survive pipeline redeployments (the next run continues from where the previous one stopped) and lets external tools inspect the subscription’s backlog:

table = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="json",
    schema=InputSchema,
    subscription_name="my-pipeline",
)

Finally, deduplication. The partition-reader recovery itself introduces no duplicates, but the topic may contain them for other reasons — a producer that retried a send (including pw.io.pulsar.write, whose delivery is at-least-once), or an upstream system that emits the same event twice. If this matters for the downstream logic — for example, the pipeline counts events — and the events carry a unique identifier, the duplicates can be removed by grouping on that identifier: every copy of an event has the same event_id, so the group produces exactly one row regardless of how many copies arrive:

class EventSchema(pw.Schema):
    event_id: str
    temperature: float
events = pw.io.pulsar.read(
    "pulsar://localhost:6650",
    "measurements",
    format="json",
    schema=EventSchema,
)
deduplicated = events.groupby(events.event_id).reduce(
    events.event_id,
    temperature=pw.reducers.earliest(events.temperature),
)

write(table, uri, topic, *, format='json', delimiter=',', key=None, value=None, headers=None, auth=None, tls_settings=None, name=None, sort_by=None)

sourceWrites data into an Apache Pulsar topic.

Every update of the table becomes a Pulsar message: an insertion carries the pathway_diff property equal to 1, and a deletion of a previously sent row carries pathway_diff equal to -1 (the attached properties and the partition key are described in the Delivery semantics section above).

• Parameters
  • table (Table) – The table to write.
  • uri (str) – The Pulsar service URI, e.g. pulsar://localhost:6650 or pulsar+ssl://my-cluster:6651 for a TLS-encrypted connection.
  • topic (str | ColumnReference) – The name of the topic to write to. It can also be a reference to a string column of the table: then each row is produced into the topic given by the value of this column, and the column itself is excluded from the message payload.
  • format (Literal['json', 'dsv', 'raw', 'plaintext']) – Format in which the message payload is produced. Can be "json", "dsv", "plaintext" or "raw". For "plaintext" and "raw", the table must consist of a single column of the string or binary type respectively, unless the value parameter points at the payload column explicitly.
  • delimiter (str) – The delimiter separating the fields, if the "dsv" format is used.
  • key (ColumnReference | None) – The column carrying the partition key of the messages. The column must be of the string or binary type. If not specified, the key is derived from the row’s primary key.
  • value (ColumnReference | None) – The column carrying the payload of the messages in the "plaintext" or "raw" formats. Can be omitted if the table has exactly one column, which then becomes the payload.
  • headers (Optional[Iterable[ColumnReference]]) – Columns to attach to every message as its properties. The values are serialized to JSON strings, because Pulsar message properties are string-to-string pairs.
  • auth (TokenAuthentication | OAuth2Authentication | None) – The authentication mechanism: TokenAuthentication, OAuth2Authentication, or None for clusters without authentication.
  • tls_settings (TLSSettings | None) – TLS connection settings. Use TLSSettings to provide the CA certificate used to verify the broker (root_cert_path) or, for development setups, to accept any broker certificate (trust_certificates=True). The connection is encrypted when the uri uses the pulsar+ssl:// scheme. Mutual TLS (client certificate) authentication is not supported.
  • name (str | None) – A unique name for the connector. If provided, this name will be used in logs and monitoring dashboards.
  • sort_by (Optional[Iterable[ColumnReference]]) – If specified, the output will be sorted in ascending order based on the values of the given columns within each minibatch. When multiple columns are provided, the corresponding value tuples will be compared lexicographically.
• Returns
  None

Example:

Suppose you want to send a stream of updates of the table t to a locally running Pulsar instance. First, create a sample table:

import pathway as pw
t = pw.debug.table_from_markdown(
    '''
    age | owner | pet
    10  | Alice | dog
    9   | Bob   | cat
    8   | Alice | cat
    '''
)

The simplest write sends every update of the table into the topic "clients" as a JSON message. Each message contains all the columns of the table, plus the pathway_time and pathway_diff properties describing the update:

pw.io.pulsar.write(t, "pulsar://localhost:6650", "clients")

If the receiving side expects a plain string instead of JSON, use the "plaintext" format and point value at the column that carries the payload:

pw.io.pulsar.write(
    t,
    "pulsar://localhost:6650",
    "clients",
    format="plaintext",
    value=t.owner,
)

The topic doesn’t have to be fixed: if it is given as a column reference, each row is produced into the topic named by that column’s value, and the column itself is excluded from the payload. Combined with key, which pins the partition of a partitioned topic (rows with equal keys keep their order), one write call can route a single table into many topics:

pw.io.pulsar.write(
    t,
    "pulsar://localhost:6650",
    topic=t.owner,
    key=t.pet,
)

Additional per-message metadata can travel in the message properties. The headers columns are attached to every message as JSON-serialized string properties, so the consumer can inspect them without parsing the payload:

pw.io.pulsar.write(
    t,
    "pulsar://localhost:6650",
    "clients",
    headers=[t.age, t.pet],
)

Finally, for a token-protected cluster pass the authentication object — the same way as in read:

pw.io.pulsar.write(
    t,
    "pulsar+ssl://my-cluster.example.com:6651",
    "clients",
    auth=pw.io.pulsar.TokenAuthentication("my-jwt-token"),
)