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. How long that cursor lives depends on the subscription_name parameter of read: a named subscription is durable and survives reconnections, while the name the connector generates when the parameter is omitted belongs to a non-durable subscription, whose cursor disappears with the connection — so a topic unload, a rebalancing or a broker restart makes such a reader resume from start_from and deliver the already processed messages again. A durable cursor narrows the repetitions rather than removing them: the broker persists it asynchronously, so a broker crash may redeliver the messages consumed since the cursor was last saved — the partition-reader mode is the one mechanism whose recovery introduces no repetitions at all. The single-consumer subscription types ("exclusive", "failover") can additionally read the compacted view of a topic (the read_compacted parameter of read), which the partition-reader mechanism does not support.

The partition-reader mechanism reads explicit positions out of the topic’s storage, so it requires a persistent topic: a non-persistent:// topic stores no messages and is rejected in the static mode, with subscription_type="reader" and with persistence. The subscription modes read such topics normally in the streaming mode.

Both mechanisms attach to the partitions the topic has when the pipeline starts, and neither picks up a partition added later: the partition readers are started once per partition, and the client library’s multi-topic consumer refreshes a topic list that already holds the physical partition names. Expanding a topic under a running pipeline is therefore reported as an error that stops the connector, rather than silently dropping everything published into the new partitions — restart the pipeline to read the expanded topic.

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, and it is resilient to broker blips: when a connection drops mid-write — a failover, a bundle unload, a rolling restart — the writer keeps a copy of every message whose broker receipt is still outstanding and republishes the unconfirmed ones with backoff for up to two minutes before failing the pipeline. A message whose receipt was lost with the connection may therefore be delivered twice, and a republication may disturb the relative order of the messages sent around the failure — outside of such recovery windows the order is preserved.

Message keys

A Pulsar message carries up to two keys, and the connectors handle both.

The partition key decides which partition of a partitioned topic the message is routed to, and serves as the key for topic compaction. Pulsar stores it as a string, so a binary key column must contain valid UTF-8. The output connector assigns the key to every message: from the column given in the key parameter of write, or, if the parameter is not set, from the row’s primary key — so the updates of one row always share a partition and keep their order. The input connector, in the "raw" and "plaintext" formats, exposes the partition key in the key column next to the payload in data, and first tries to use it as the row’s primary key, autogenerating one for the messages published without a partition key — the way the Kafka connector treats message keys. The row identity therefore follows the partition key rather than the message: several messages sharing a partition key produce several rows with the same primary key, which the operations relying on key uniqueness reject. This fits the topics whose keys identify individual events; when the keys repeat and every message must stay a separate row, the autogenerate_key parameter of read switches to autogenerated primary keys.

The ordering key is a separate, optional key used by the "key_shared" subscriptions: when present, the broker distributes the messages between the consumers by hashing it instead of the partition key, so the messages of one entity stay ordered on one consumer while the partition routing still follows the partition key. The ordering_key parameter of write sets it from a column; on the reading side its value is available in the _metadata column (with_metadata=True).

Compression

The output connector can compress the message payloads: the compression parameter of write accepts "lz4", "zlib" and "zstd". Compression is transparent to the consumers — the codec travels in the message metadata, so any Pulsar client, including pw.io.pulsar.read, decompresses the messages automatically, with no matching setting on the reading side.

Snappy is not supported: the underlying client library uses the snappy frame format, while the other Pulsar clients use the raw block format. Messages compressed this way would be unreadable for every non-Rust consumer of the topic, so the codec is rejected; reading a topic with snappy-compressed messages fails with a decompression error.

Timestamps

A Pulsar message carries two timestamps, both in milliseconds since the UNIX epoch. The publish time is assigned by the broker when the message is accepted; it always exists, and it is the timestamp the start_from parameter of read filters on. The event time is optional and assigned by the producer, describing when the underlying event happened. The event_time parameter of write sets it: either from a column of the table (an integer in milliseconds or a UTC datetime), or, with pw.io.ENGINE_TIME, from the engine time of the update — the same value the messages carry in the pathway_time property, placed into the native field so the consumers read it without parsing the properties. When neither is needed, the field is left unset. On the reading side both timestamps are available in the _metadata column (with_metadata=True).

Schema registry

Pulsar’s schema registry is a part of the broker: a topic carries a versioned schema, the producers declare theirs on connect (validated against the namespace compatibility policy), and every message is stamped with the version it was produced under. The connectors integrate with it on three levels.

Writing. write with format="avro" derives an Avro record schema from the schema of the table, declares it to the registry and publishes the rows as bare Avro binary datums — no container header and no schema identifier inside the body, which is how Pulsar transports typed payloads. Any schema-aware Pulsar consumer can then decode the topic. An incompatible schema is rejected by the broker per its compatibility policy. Optional columns become ["null", T] unions with a null default, so adding an optional column remains a backward-compatible schema change. The registered schema is a public contract of the topic, so the columns designated as the service inputs of write — the dynamic topic, the partition and ordering keys, the event time, the headers — are not a part of the record.

Reading. read with format="avro" obtains the writer schema of every message from the registry by the stamped version (cached per version), decodes the datum with it and projects the decoded fields onto the table columns by name. A topic whose schema evolved stays readable: a column the writer schema does not know takes its declared default, and an optional column without one reads as None. A transient failure to reach the registry is retried with backoff for as long as it lasts, instead of condemning the message: the reader has already acknowledged it, so the read stalls (backpressure) until the registry answers — a registry outage pauses the pipeline rather than losing rows. The registry’s definitive refusals (no such version, a non-AVRO version, rejected permissions) are per-row errors and are remembered per version. The lookups run through short-lived probe consumers with generated subscription names; in the deployments where a role is authorized for specific subscription names only, that role must also permit these probes — otherwise every lookup fails with an authorization error, for the deduction and the format="avro" reads alike. A message that carries no schema version is a per-row error — unless the table schema is explicit, in which case it doubles as the decoding schema for such messages, making bare-datum topics without a registered schema readable.

Schema deduction. When the schema parameter of read is omitted for the "avro" and "json" formats, the columns of the table are deduced from the topic’s current registry schema at pipeline construction time, together with the field defaults and documentation strings. The registry type of the topic must agree with the requested format — an AVRO-typed topic read with format="json" (or the other way around) is rejected up front. An explicit schema always overrides the deduction. Multi-process runs and persistent pipelines require an explicit schema: the deduction runs anew at every start, so a schema change between the starts would make the processes (or the restarted pipeline and its snapshot) disagree about the table layout.

The table below explains how the columns are written into the Avro record that the connector registers and publishes. The same conversions are used in reverse when reading a topic written by Pathway, so a write-read roundtrip preserves every type.

Pathway types conversion into Avro

Reading topics produced by other tools

The topics written by other producers commonly use Avro types Pathway never writes itself. pw.io.pulsar.read accepts the following extra Avro types and projects them onto the Pathway type declared in the schema (or deduced from the registry):

Additional Avro-to-Pathway conversions (read only)

Two features of other Pulsar producers are not supported on the reading side, and the reader fails with an error naming the cause when it meets such a message rather than corrupt the data silently: chunking (a payload above the broker’s maxMessageSize split into several messages — the reader cannot reassemble the fragments) and end-to-end encryption (the reader has no decryption keys and would otherwise deliver ciphertext). A start_from timestamp above such an era of the topic skips past it.

For the JSON-typed topics the deduction follows the JSON parser instead of the table above: the Avro logical time types deduce into plain int columns, holding the raw numbers the JSON payloads carry. The JSON parser reads datetimes as formatted strings and durations as nanosecond counts, so applying the Avro units there would corrupt the values.

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, read_compacted=False, autocommit_duration_ms=1500, json_field_paths=None, autogenerate_key=False, with_metadata=False, 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 four formats supported: "plaintext", "raw", "json" and "avro".

For the "raw" format, the partition key and the payload are read as raw bytes and added to the table as they are. In the "plaintext" format they are decoded from UTF-8 and stored as plain text. In both cases the table has two columns: "key" with the partition key of the message (or None if the message was published without one) and "data" with the payload. The connector first tries to use the partition key of the message as the primary key of the row, and autogenerates one when the message has no partition key. Note that this makes the row identity follow the partition key rather than the message: several messages sharing a partition key produce several rows with the same primary key, which the operations relying on key uniqueness reject. Set autogenerate_key=True to give every message its own row instead.

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.

Compressed messages (lz4, zlib, zstd) are decompressed automatically: the codec travels in the message metadata, so no configuration is needed on the reading side. Snappy-compressed topics are not supported: reading such messages fails with a decompression error.

• 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. A non-persistent:// topic is accepted only in the streaming mode with a subscription type (see the topic-persistence notes in the overview above).
  • schema (type[Schema] | None) – Schema of the resulting table. For the "json" and "avro" formats it may be omitted: the columns are then deduced from the topic’s registry schema (see the Schema registry section above for the mechanics and the restrictions). An explicit schema always overrides the deduction; a topic with no registered schema, a multi-process run and a persistent pipeline require one.
  • format (Literal['plaintext', 'raw', 'json', 'avro']) – Format of the incoming messages: "plaintext", "raw", "json" or "avro". With "avro", the payloads are bare Avro binary datums decoded through the broker’s schema registry — the Schema registry section above describes the mechanics, the schema-evolution behavior and the type conversions. A message produced under a non-AVRO registry version is reported as a per-row error without stopping the pipeline; so is a message produced without a schema — unless the table schema is explicit, in which case it doubles as the decoding schema for such messages.
  • 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 — but a non-durable cursor only lives as long as the connection, so whenever the broker drops it (a topic unload, a rebalancing, a broker restart) the reading resumes from start_from and the messages processed so far are delivered again (the overview above compares the recovery guarantees of the mechanisms). Multi-process runs of the "shared" and "key_shared" types require an explicit name as well, 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.
  • read_compacted (bool) – If True, the subscription reads the compacted view of the topic: for the part of the topic that has been compacted, only the latest message of every partition key is delivered, followed by the messages published after the compaction horizon as usual. Pulsar restricts compacted reads to the single-consumer subscriptions, so this requires the "exclusive" or "failover" subscription type and the "streaming" mode; the partition-reader mechanism (the static mode, persistence, subscription_type="reader") does not support it.
  • 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).
  • autogenerate_key (bool) – If True, Pathway autogenerates a unique primary key for every message. Otherwise it first tries to use the partition key of the message, and autogenerates the key only for the messages published without one — so the messages sharing a partition key produce rows with the same primary key. Use True when the partition keys repeat and every message must stay a separate row. This parameter is only used with the "raw" and "plaintext" formats.
  • with_metadata (bool) – When set to True, the connector adds an additional column named _metadata to the table, a JSON field describing the Pulsar message the row came from: topic (the physical topic, including the -partition-N suffix for partitioned topics), partition (-1 for a non-partitioned topic), ledger_id, entry_id and batch_index (the components of the message id; the batch index is -1 for non-batched messages), publish_time_millis (the broker-assigned publish timestamp, in milliseconds since the UNIX epoch), event_time_millis (the producer-assigned event timestamp, or null if the producer didn’t set one), producer_name, ordering_key (base64-encoded, or null), schema_version (the registry version the message was produced under, or null for the messages produced without a schema) and properties (the user-defined message properties, a string-to-string map).
  • 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 (the default) for clusters without authentication — see the Authentication section above.
  • tls_settings (TLSSettings | None) – TLS connection settings, described in the Authentication section above. None (the default) leaves the client with its default certificate verification.
  • 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 the data column as a UTF-8 string, with the partition key of the message in the key column ("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, and neither does a named (durable) subscription. An auto-generated subscription does: it is non-durable, so a broker-side reconnection restarts it from start_from. The topic may also contain duplicates for reasons of its own — 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, ordering_key=None, event_time=None, deliver_at=None, deliver_after=None, value=None, headers=None, compression=None, producer_name=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. In the "avro" format the topic column — like the other service columns — is not a part of the payload; the other formats include every column of the table in the payload.
  • format (Literal['json', 'dsv', 'raw', 'plaintext', 'avro']) – Format in which the message payload is produced. Can be "json", "dsv", "plaintext", "raw" or "avro". 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. For "avro", the payload is the Avro binary encoding of the row, and the derived schema is declared to the broker’s schema registry — see the Schema registry section above for the mechanics and the type conversions. The columns designated as the service inputs — topic, key, ordering_key, event_time, deliver_at, deliver_after, headers — stay out of the record; duplicate a column under another name with table.select(...) if it must also be a part of the payload.
  • 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; a binary key must be valid UTF-8, because Pulsar stores partition keys as strings. If not specified, the key is derived from the row’s primary key.
  • ordering_key (ColumnReference | None) – The column carrying the ordering key of the messages, of the string or binary type. The ordering key is what a key_shared subscription hashes when distributing the messages between its consumers: setting it keeps the messages of one entity in order on one consumer while the partition routing still follows key. If not specified, the ordering key is left unset and key_shared subscriptions fall back to the partition key.
  • event_time (ColumnReference | EngineTimeMarker | None) – Where the event_time of the messages comes from. A column reference takes the value from that column of the table: an integer (milliseconds since the UNIX epoch) or a UTC datetime. The pw.io.ENGINE_TIME marker uses the engine (minibatch) time of the update — the same UNIX timestamp the messages carry in the pathway_time property, but placed into the native event_time field, so the consumers read it without parsing the properties. If not set (the default), the event_time field is left unset and the messages only carry the broker-assigned publish time. Note that for the static tables created with pw.debug utilities the engine time is a small logical counter rather than a wall-clock timestamp; data read through pw.io.* connectors always carries wall-clock engine times, in both the streaming and the static modes.
  • deliver_at (ColumnReference | None) – The column carrying the delivery time of the messages, for the delayed delivery — an integer (milliseconds since the UNIX epoch) or a UTC datetime. The broker accepts and stores such a message right away, but hands it to the consumers only once the delivery time comes; a delivery time in the past delivers the message immediately. Mutually exclusive with deliver_after. See the Delayed delivery section below for what the consumers must look like to honor the schedule, and for the configurations that are rejected.
  • deliver_after (ColumnReference | timedelta | None) – The delay of the delivery, counted from the moment the message is published: either a constant datetime.timedelta applied to every message, or a column reference holding the delay of each row — a duration or an integer number of milliseconds. Mutually exclusive with deliver_at; the same caveats apply.
  • 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.
  • compression (Optional[Literal['lz4', 'zlib', 'zstd']]) – The codec the message payloads are compressed with: "lz4", "zlib", "zstd", or None (the default) to send them uncompressed. Compression is transparent to the consumers: the codec travels in the message metadata, so any Pulsar client — including pw.io.pulsar.read — decompresses the messages automatically, with no matching setting on the reading side. Snappy is not supported: the underlying client library frames it differently from the other Pulsar clients, so such messages would be unreadable outside Pathway.
  • producer_name (str | None) – The name the producers register themselves under on the broker, visible in the topic stats and logs — useful to identify the pipeline among the writers of a topic. Pulsar requires producer names to be unique within a topic and every Pathway worker runs its own producer, so the worker index is appended to the name (e.g. my-pipeline-0). If not set, the broker assigns a generated name.
  • auth (TokenAuthentication | OAuth2Authentication | None) – The authentication mechanism: TokenAuthentication, OAuth2Authentication, or None (the default) for clusters without authentication — see the Authentication section above.
  • tls_settings (TLSSettings | None) – TLS connection settings, described in the Authentication section above. None (the default) leaves the client with its default certificate verification.
  • 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

Delayed delivery:

deliver_at and deliver_after schedule the messages for a later delivery. The schedule is honored only by the consumers of the "shared" and "key_shared" subscriptions — the "exclusive" and "failover" subscriptions and the partition readers receive the messages as soon as they are published, as the Pulsar protocol defines. pw.io.pulsar.read therefore observes the schedule only with subscription_type="shared" or "key_shared" in the streaming mode; its default partition-reader mechanism, the persistence-enabled pipelines and the static mode read the scheduled messages immediately. The broker must run with the delayed delivery enabled (delayedDeliveryEnabled, the default), otherwise it ignores the schedule; it releases the messages with the precision of its delayed-delivery tick (delayedDeliveryTickTimeMillis, a second by default). Every update of the table follows the schedule of its row, the deletions included.

The configurations that cannot work are rejected instead of publishing messages that would be delivered immediately: deliver_at and deliver_after together, a timezone-naive datetime column, a column of another type, a negative delay or delivery time (at the construction or when the row is written), and a non-persistent:// topic, which stores no messages and dispatches everything at once. The scheduled messages are published one by one rather than in the producer batches, so the write throughput with a schedule is lower than without one.

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 (see the topic parameter for whether the column is part of 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],
)

To reduce the network traffic and the storage used by the topic, the payloads can be compressed. The consumers decompress transparently, so no setting is needed on the reading side:

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

A message can be scheduled for a later delivery. With a constant delay every message is handed to the "shared" and "key_shared" consumers a minute after it is published:

import datetime
pw.io.pulsar.write(
    t,
    "pulsar://localhost:6650",
    "clients",
    deliver_after=datetime.timedelta(minutes=1),
)

The schedule can also come from the data — for example, a reminder is delivered at the time stored in its row, given either as an integer number of milliseconds since the UNIX epoch or as a UTC datetime:

reminders = pw.debug.table_from_markdown(
    '''
    text        | remind_at_ms
    water-plants | 1893456000000
    call-alice   | 1893542400000
    '''
)
pw.io.pulsar.write(
    reminders,
    "pulsar://localhost:6650",
    "reminders",
    deliver_at=reminders.remind_at_ms,
)

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"),
)