Pathway User Guide

Pathway is a Python framework for building real-time data processing pipelines. It provides a high-level, declarative API for defining streaming computations that automatically update as new data arrives. Under the hood, a high-performance Rust engine (based on differential dataflow) handles the heavy lifting, while you write idiomatic Python.

This guide covers everything you need to install, configure, and use Pathway effectively — from simple ETL scripts to complex real-time AI/RAG applications.

Note

Pathway is licensed under the Business Source License 1.1 (BUSL-1.1). Some advanced features (persistence, multi-process scaling) require a license key. Community usage for development, testing, and small-scale production is freely available.

1. Quick Start

Your First Pipeline in 60 Seconds

pip install -U pathway

import pathway as pw

# Read CSV data reactively
table = pw.io.csv.read("./input_data/", schema=pw.schema_from_types(name=str, value=int))

# Transform
filtered = table.filter(pw.this.value > 100)
result = filtered.select(pw.this.name, doubled=pw.this.value * 2)

# Write output
pw.io.csv.write(result, "./output_data/result.csv")

# Execute the dataflow
pw.run()

Save this as pipeline.py and run:

python pipeline.py

Pathway monitors the ./input_data/ directory. Drop a CSV file in, and the output updates automatically. Add more files — the results reflect the entire dataset incrementally.

Key Mental Model

1. Define your data sources (inputs)
2. Transform using table operations (filter, join, groupby, windowby)
3. Write to outputs (files, databases, message queues)
4. Run — pw.run() starts the engine; it processes data continuously until stopped

Important

pw.run() is blocking. It starts the Rust engine and processes data indefinitely (for streaming sources) or until all batch data is consumed. Place it as the last call in your script.

2. Installation

Requirements

Standard Install

pip install -U pathway

With Optional Dependencies

# For all connectors and xpacks (LLM, spatial, etc.)
pip install -U "pathway[all]"

# For LLM/RAG features specifically
pip install -U "pathway[xpack-llm]"

Verify Installation

python -c "import pathway as pw; print(pw.__version__)"

Docker

docker run -it pathwaycom/pathway:latest python -c "import pathway as pw; print('OK')"

Tip

If you encounter issues with native dependencies on macOS or Windows, use Docker or WSL2. Pathway's Rust engine compiles against the CPython ABI (abi3-py310), so any Python 3.10+ works without version-specific wheels.

3. Configuration

License Key

Advanced features (multi-process scaling, persistence, some enterprise connectors) require a license key:

import pathway as pw
pw.set_license_key("your-key-here")

Or via environment variable:

export PATHWAY_LICENSE_KEY="your-key-here"

Environment Variables

YAML Pipeline Configuration

Pathway supports defining entire pipelines declaratively in YAML:

# pipeline.yaml
schema: !pw.schema_from_types
  name: str
  value: int
  timestamp: int

input: !pw.io.kafka.read
  rdkafka_settings:
    bootstrap.servers: $KAFKA_BOOTSTRAP_SERVERS
    group.id: "pathway-consumer"
  topic: "events"
  schema: !ref schema
  format: "json"

output: !pw.io.postgres.write
  connection_string: $POSTGRES_DSN
  table_name: "processed_events"

persistence: !pw.persistence.Config
  backend: !pw.persistence.Backend.filesystem
    path: "./state"

Load and run:

import pathway as pw
pw.load_yaml("pipeline.yaml")
pw.run()

Environment variables are interpolated using $VAR_NAME syntax within the YAML.

Note

For complete configuration reference including persistence backends, monitoring setup, and multi-node deployment parameters, see the Admin Guide.

4. Core Usage

Tables and Schemas

Tables are Pathway's core abstraction — immutable, versioned collections of rows that update in real time.

import pathway as pw

# Define a schema
class SensorReading(pw.Schema):
    sensor_id: str
    temperature: float
    timestamp: int

# Read with explicit schema
readings = pw.io.csv.read("./sensors/", schema=SensorReading)

Or use inline schema:

readings = pw.io.csv.read(
    "./sensors/",
    schema=pw.schema_from_types(sensor_id=str, temperature=float, timestamp=int)
)

Column References

Pathway uses special references for column access:

# Select and rename
result = table.select(
    name=pw.this.first_name + " " + pw.this.last_name,
    score=pw.this.value * 2
)

Filtering

# Simple filter
high_temp = readings.filter(pw.this.temperature > 80.0)

# Compound conditions
alerts = readings.filter(
    (pw.this.temperature > 95.0) & (pw.this.sensor_id != "test-sensor")
)

Groupby and Aggregation

# Group and aggregate
stats = readings.groupby(pw.this.sensor_id).reduce(
    sensor_id=pw.this.sensor_id,
    avg_temp=pw.reducers.avg(pw.this.temperature),
    max_temp=pw.reducers.max(pw.this.temperature),
    count=pw.reducers.count(),
)

Available reducers include: pw.reducers.sum, pw.reducers.avg, pw.reducers.min, pw.reducers.max, pw.reducers.count, pw.reducers.sorted_tuple, pw.reducers.tuple, pw.reducers.any, pw.reducers.argmin, pw.reducers.argmax, pw.reducers.earliest, pw.reducers.latest, pw.reducers.unique.

Joins

# Inner join
joined = orders.join(customers, pw.left.customer_id == pw.right.id).select(
    order_id=pw.left.id,
    customer_name=pw.right.name,
    amount=pw.left.amount,
)

# Left join
joined = orders.join_left(customers, pw.left.customer_id == pw.right.id).select(
    order_id=pw.left.id,
    customer_name=pw.right.name,
    amount=pw.left.amount,
)

Adding and Renaming Columns

# Add columns
enriched = table.with_columns(
    full_name=pw.this.first + " " + pw.this.last,
    is_premium=pw.this.total_spend > 1000,
)

# Rename
renamed = table.rename({"old_name": "new_name"})

# Cast types
casted = table.cast({"value": float, "count": int})

Flatten

# Flatten a column containing lists
exploded = table.flatten(pw.this.tags)

SQL Interface

result = pw.sql(
    """
    SELECT sensor_id, AVG(temperature) as avg_temp
    FROM readings
    WHERE temperature > 50
    GROUP BY sensor_id
    """,
    readings=readings
)

User-Defined Functions (UDFs)

# Simple UDF
@pw.udf
def normalize(value: float, min_val: float, max_val: float) -> float:
    return (value - min_val) / (max_val - min_val)

result = table.select(normalized=normalize(pw.this.value, 0.0, 100.0))

# Async UDF (for I/O-bound operations)
@pw.udf(executor=pw.udfs.async_executor())
async def fetch_enrichment(key: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"http://api.example.com/{key}") as resp:
            return await resp.text()

Fixed-Point Iteration

# Iterative computation (e.g., graph algorithms)
result = pw.iterate(
    lambda table: table.join(edges, pw.left.node == pw.right.src).select(
        node=pw.right.dst,
        distance=pw.left.distance + pw.right.weight,
    ),
    initial_table
)

Interactive / Jupyter Mode

import pathway as pw
pw.enable_interactive_mode()

# Now tables evaluate eagerly for exploration
table = pw.io.csv.read("./data/", schema=MySchema)
table.show()   # Displays current state
table.plot()   # Visualize (if applicable)

Tip

Interactive mode is for development and exploration only. In production pipelines, use pw.run() for full streaming execution.

5. Common Tasks

Task: Stream from Kafka, Write to PostgreSQL

import pathway as pw

class EventSchema(pw.Schema):
    user_id: str
    event_type: str
    timestamp: int
    payload: str

events = pw.io.kafka.read(
    rdkafka_settings={
        "bootstrap.servers": "localhost:9092",
        "group.id": "pathway-pipeline",
    },
    topic="user-events",
    schema=EventSchema,
    format="json",
)

# Filter and transform
important = events.filter(pw.this.event_type != "heartbeat")
processed = important.select(
    pw.this.user_id,
    pw.this.event_type,
    pw.this.timestamp,
    event_length=pw.apply(len, pw.this.payload),
)

pw.io.postgres.write(
    processed,
    connection_string="postgresql://user:pass@localhost:5432/mydb",
    table_name="processed_events",
)

pw.run()

Task: Monitor a Directory for New Files

import pathway as pw

# Monitors directory recursively; processes new/modified files
logs = pw.io.fs.read(
    "./incoming_logs/",
    format="json",
    schema=pw.schema_from_types(level=str, message=str, ts=int),
    mode="streaming",
)

errors = logs.filter(pw.this.level == "ERROR")
pw.io.jsonlines.write(errors, "./alerts/errors.jsonl")
pw.run()

Task: Real-Time Aggregation with Tumbling Windows

import pathway as pw
from pathway.stdlib.temporal import tumbling

class TradeSchema(pw.Schema):
    symbol: str
    price: float
    volume: int
    timestamp: int

trades = pw.io.kafka.read(
    rdkafka_settings={"bootstrap.servers": "localhost:9092", "group.id": "agg"},
    topic="trades",
    schema=TradeSchema,
    format="json",
)

# 1-minute tumbling windows
windowed = trades.windowby(
    pw.this.timestamp,
    window=tumbling(duration=60),
    instance=pw.this.symbol,
).reduce(
    symbol=pw.this.symbol,
    vwap=pw.reducers.sum(pw.this.price * pw.this.volume) / pw.reducers.sum(pw.this.volume),
    total_volume=pw.reducers.sum(pw.this.volume),
    trade_count=pw.reducers.count(),
)

pw.io.jsonlines.write(windowed, "./output/vwap.jsonl")
pw.run()

Task: Sliding Window Aggregation

from pathway.stdlib.temporal import sliding

windowed = sensor_data.windowby(
    pw.this.timestamp,
    window=sliding(duration=300, hop=60),  # 5-min window, 1-min hop
    instance=pw.this.sensor_id,
).reduce(
    sensor_id=pw.this.sensor_id,
    avg_reading=pw.reducers.avg(pw.this.value),
    max_reading=pw.reducers.max(pw.this.value),
)

Task: Session Windows

from pathway.stdlib.temporal import session

# Group events within 30-second gaps
sessions = clickstream.windowby(
    pw.this.timestamp,
    window=session(max_gap=30),
    instance=pw.this.user_id,
).reduce(
    user_id=pw.this.user_id,
    click_count=pw.reducers.count(),
    session_duration=pw.reducers.max(pw.this.timestamp) - pw.reducers.min(pw.this.timestamp),
)

Task: Temporal Joins (As-Of Join)

Join two streams by time proximity — match each row from the left table with the most recent row from the right table:

# Match each trade with the latest quote
enriched = trades.asof_join(
    quotes,
    pw.left.timestamp,
    pw.right.timestamp,
    pw.left.symbol == pw.right.symbol,
).select(
    symbol=pw.left.symbol,
    trade_price=pw.left.price,
    bid=pw.right.bid,
    ask=pw.right.ask,
)

Variants: asof_join_left, asof_join_right, asof_join_outer.

Task: Interval Join

Join rows where timestamps fall within a specified interval of each other:

# Match orders with shipments within 7 days
matched = orders.interval_join(
    shipments,
    pw.left.order_time,
    pw.right.ship_time,
    pw.temporal.interval(-1, 7 * 86400),  # -1s to +7 days
    pw.left.order_id == pw.right.order_id,
).select(
    order_id=pw.left.order_id,
    order_time=pw.left.order_time,
    ship_time=pw.right.ship_time,
)

Variants: interval_join_inner, interval_join_left, interval_join_right, interval_join_outer.

Task: Window Join

result = left_table.window_join(
    right_table,
    pw.left.timestamp,
    pw.right.timestamp,
    pw.temporal.tumbling(duration=60),
    pw.left.key == pw.right.key,
).select(
    key=pw.left.key,
    left_val=pw.left.value,
    right_val=pw.right.value,
)

Variants: window_join_inner, window_join_left, window_join_right, window_join_outer.

Task: Inactivity Detection

Detect keys that haven't received updates within a threshold:

inactive = sensor_data.inactivity_detection(
    pw.this.timestamp,
    inactivity_threshold=300,  # 5 minutes
    instance=pw.this.sensor_id,
)

Task: Multi-Source Merge

import pathway as pw

csv_data = pw.io.csv.read("./batch/", schema=MySchema)
kafka_data = pw.io.kafka.read(
    rdkafka_settings={"bootstrap.servers": "broker:9092", "group.id": "merge"},
    topic="live-events",
    schema=MySchema,
    format="json",
)

# Union tables from different sources
combined = pw.Table.concat_reindex(csv_data, kafka_data)

pw.io.jsonlines.write(combined, "./merged_output.jsonl")
pw.run()

Task: Build a Real-Time RAG Pipeline

import pathway as pw
from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.vector_store import VectorStoreServer

# Watch a folder for documents
documents = pw.io.fs.read(
    "./knowledge_base/",
    format="binary",
    with_metadata=True,
)

# Split into chunks and embed
splitter = TokenCountSplitter(min_tokens=100, max_tokens=400)
embedder = OpenAIEmbedder(model="text-embedding-3-small")

vector_server = VectorStoreServer(
    documents,
    embedder=embedder,
    splitter=splitter,
)

# Serve retrieval endpoint
vector_server.run_server(host="0.0.0.0", port=8765)

Query the running server:

curl -X POST http://localhost:8765/v1/retrieve \
  -H "Content-Type: application/json" \
  -d '{"query": "How do I configure logging?", "k": 5}'

The /v1/statistics endpoint returns index metadata:

curl http://localhost:8765/v1/statistics

Important

The VectorStoreServer updates its index in real time. Drop new documents into the watched folder and they become searchable within seconds — no manual re-indexing required.

Task: Custom Python Connector (Source)

import pathway as pw

class MySource(pw.io.python.ConnectorSubject):
    def run(self):
        # Called once; emit rows via self.next()
        for i in range(100):
            self.next(sensor_id=f"s{i}", value=i * 1.5)
        # Signal completion for batch mode
        self.close()

table = pw.io.python.read(
    MySource(),
    schema=pw.schema_from_types(sensor_id=str, value=float),
)

Task: Custom Python Connector (Sink / Observer)

class MySink(pw.io.python.ConnectorObserver):
    def on_change(self, key, row, time, is_addition):
        if is_addition:
            print(f"New row: {row}")

pw.io.python.write(result_table, MySink())
pw.run()

Task: Write to Multiple Outputs

# Same table, multiple destinations
pw.io.csv.write(processed, "./output/data.csv")
pw.io.kafka.write(
    processed,
    rdkafka_settings={"bootstrap.servers": "broker:9092"},
    topic_name="processed-output",
    format="json",
)
pw.io.postgres.write(
    processed,
    connection_string="postgresql://user:pass@db:5432/analytics",
    table_name="results",
)
pw.run()  # Single run() drives all outputs

Task: S3 Input/Output

data = pw.io.s3.read(
    "s3://my-bucket/input/",
    aws_s3_settings=pw.io.s3.AwsS3Settings(
        bucket_name="my-bucket",
        region="us-east-1",
    ),
    format="json",
    schema=MySchema,
)

pw.io.s3.write(
    processed,
    "s3://my-bucket/output/",
    aws_s3_settings=pw.io.s3.AwsS3Settings(
        bucket_name="my-bucket",
        region="us-east-1",
    ),
    format="json",
)

Task: MongoDB Read/Write

data = pw.io.mongodb.read(
    connection_string="mongodb://localhost:27017",
    database="mydb",
    collection="events",
    schema=EventSchema,
)

pw.io.mongodb.write(
    results,
    connection_string="mongodb://localhost:27017",
    database="mydb",
    collection="processed",
)

Task: HTTP Server Connector (Receive webhooks)

# Accept data via HTTP POST
webhook_data = pw.io.http.read(
    host="0.0.0.0",
    port=8080,
    schema=pw.schema_from_types(event=str, data=str),
    format="json",
)

Task: Elasticsearch Output

pw.io.elasticsearch.write(
    enriched_data,
    connection_string="http://localhost:9200",
    index_name="processed-events",
)

Task: Delta Lake / Iceberg Output

# Write to Delta Lake
pw.io.deltalake.write(processed, "./delta_table/")

# Write to Iceberg
pw.io.iceberg.write(
    processed,
    catalog_uri="http://localhost:8181",
    namespace="analytics",
    table_name="events",
)

Task: Diff / Change Detection

# Get additions and deletions as they happen
changes = table.diff(pw.this.value)

Task: Add Update Timestamps

# Track when each row was last updated
timestamped = table.add_update_timestamp_utc(pw.this.updated_at)

Task: Demo / Testing Streams

# Generate fake bounded data for testing
demo_table = pw.demo.range_stream(nb_rows=1000)

# Generate unbounded stream with configurable rate
live_demo = pw.demo.noisy_linear_stream(nb_rows=10000, input_rate=100)

6. Troubleshooting

Installation Issues

Problem: pip install pathway fails with compilation errors Pathway ships pre-built wheels for common platforms. If pip falls back to source compilation, ensure you have Rust 1.93+ and a C compiler. Prefer using a supported platform (Linux x86_64 with glibc 2.31+).

Problem: Import error — cannot find libpq or similar shared library errors Some connector modules (PostgreSQL) depend on system libraries. Install them:

# Ubuntu/Debian
sudo apt-get install libpq-dev

# macOS
brew install libpq

Problem: ModuleNotFoundError: No module named 'pathway' after install Verify you're using the same Python environment:

which python
python -m pip show pathway

Problem: Pathway does not work on native Windows Use WSL2 or Docker. The Rust engine targets POSIX systems.

Problem: Version mismatch between Python and wheel Pathway requires Python 3.10+. Check with python --version. Upgrade if necessary.

Runtime Issues

Problem: pw.run() hangs and produces no output

• For filesystem connectors: ensure the input directory exists and contains files matching the expected format
• For streaming connectors: verify the source (Kafka, etc.) is reachable and producing data
• Check that your pipeline has at least one output connector; without an output, the engine has nothing to materialize

Problem: pw.run() exits immediately with no output

• For batch sources (CSV files already present), this is normal — processing completes instantly
• Verify output files exist at the specified path
• If expecting streaming behavior, ensure the connector is configured in streaming mode

Problem: Pipeline processes data but output file is empty Pathway uses a transactional output model. For CSV output, data may be buffered. Check that:

• The pipeline has actually received matching rows after filtering
• The output path is writable

Problem: High memory usage with large datasets

• Pathway's Rust engine manages memory via differential dataflow. Memory grows with the number of distinct keys maintained in state
• For windowed aggregations, ensure windows have bounded duration — unbounded windows accumulate state forever
• Use pathway spawn -t N to parallelize across threads

Problem: RuntimeError: Cannot run two dataflows simultaneously Only one pw.run() can be active per process. Structure your code so all tables and outputs are defined before calling pw.run() once.

Problem: Schema mismatch — TypeError or KeyError on column access Verify your schema matches the actual data. Column names are case-sensitive. Use pw.schema_from_types() for quick prototyping.

Problem: Slow startup / "Compiling dataflow..." takes too long The first run compiles the dataflow graph. For complex pipelines with many joins, this is expected. Subsequent data processing is fast.

Problem: ConnectionRefusedError when connecting to Kafka/Postgres/etc.

• Verify the service is running and accessible from your host
• Check firewall rules and connection strings
• For Kafka: verify bootstrap.servers and group.id are correct

Problem: Processing 1M rows is unexpectedly slow This is a known issue (#21). For large batch workloads:

• Increase thread count: pathway spawn -t 4 your_script.py
• Profile whether the bottleneck is I/O or computation
• Ensure you're not doing row-by-row UDF calls for operations that could be vectorized

Problem: PATHWAY_LICENSE_KEY not recognized Set it before importing pathway, or use pw.set_license_key() at the top of your script, before defining any tables.

Problem: UDF raises exception — pipeline crashes UDF exceptions propagate and terminate the pipeline by default. Wrap your UDF logic in try/except and return a sentinel value for error cases:

@pw.udf
def safe_parse(data: str) -> str:
    try:
        return json.loads(data)["field"]
    except (json.JSONDecodeError, KeyError):
        return ""

Problem: setuptools dependency conflict Known issue (#190). If you encounter this, pin setuptools or install pathway in an isolated virtual environment:

python -m venv .venv && source .venv/bin/activate
pip install -U pathway

Problem: Nightly builds require external libpq Known issue (#199). Stable releases bundle dependencies; nightly builds may not. Stick to stable releases for production.

Problem: OnChangeCallbackAsync not found in public API Known issue (#193). Use pw.io.python.ConnectorObserver instead for observing output changes.

Connector-Specific Issues

Problem: Kafka consumer not receiving messages

• Verify group.id is unique for this consumer
• Check topic exists: kafka-topics --list --bootstrap-server localhost:9092
• Ensure messages are in the expected format (json, raw, etc.)
• Check consumer group offset: new groups start at latest offset by default

Problem: S3 connector authentication failure Ensure AWS credentials are available via environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY), IAM role, or AWS CLI configuration.

Problem: PostgreSQL write fails with permission error The database user needs INSERT and CREATE TABLE privileges on the target schema.

Problem: Filesystem connector misses files

• Files must be fully written before Pathway detects them (use atomic move/rename)
• Hidden files (starting with .) may be skipped
• Ensure mode="streaming" for continuous monitoring

Problem: HTTP connector returns 400 errors Verify the POST body matches the declared schema exactly. Extra or missing fields cause rejection.

Problem: MongoDB connector slow on large collections Use change streams (streaming mode) rather than polling the entire collection. Ensure the MongoDB replica set is configured for change stream support.

Problem: MQTT/NATS connector disconnects These protocols require persistent connections. Check network stability, broker timeouts, and keep-alive settings.

Multi-Process / Scaling Issues

Problem: pathway spawn -n 4 fails with port conflicts The --first-port flag sets the base port for inter-process communication. Ensure ports first_port through first_port + n - 1 are available:

pathway spawn -n 4 --first-port 11000 my_pipeline.py

Problem: Multi-machine deployment not distributing work Use --addresses to specify all nodes:

# On machine 1:
pathway spawn -n 2 --addresses "machine1:10000,machine1:10001,machine2:10000,machine2:10001" -pi 0 pipeline.py

# On machine 2:
pathway spawn -n 2 --addresses "machine1:10000,machine1:10001,machine2:10000,machine2:10001" -pi 2 pipeline.py

Problem: State lost after restart Enable persistence:

export PATHWAY_PERSISTENCE_MODE="speedrun"
export PATHWAY_SNAPSHOT_ACCESS="record"

This records state snapshots that can be replayed on restart.

Problem: pathway replay doesn't reproduce original results Ensure --record-path points to the correct recording directory. Use --mode to control replay speed.

Monitoring Issues

Problem: Dashboard shows no metrics Ensure PATHWAY_MONITORING_HTTP_PORT is set before pw.run():

export PATHWAY_MONITORING_HTTP_PORT=20000
python pipeline.py

Then access at http://localhost:20000.

Problem: pathway web_dashboard shows blank page Verify the --detailed-metrics-dir points to where the engine writes metrics (default: current directory). Check that the pipeline is actually running and producing metrics.

7. FAQ

General

Q: What makes Pathway different from Spark Structured Streaming or Flink? Pathway is Python-native — no JVM, no cluster manager, no YARN. You write standard Python, and the Rust engine handles incremental computation. It's designed for single-machine or small-cluster deployments with microsecond-level incremental updates.

Q: Can I use Pathway for batch processing? Yes. Point it at static files (CSV, JSON, Parquet) and pw.run() processes everything and exits. The same pipeline code works for both batch and streaming without modification.

Q: Is Pathway single-threaded? By default, yes (1 thread). Scale with pathway spawn -t N for multi-threaded or -n N for multi-process execution.

Q: What's the difference between pw.run() and pw.run_all()? pw.run() executes the current computation graph. pw.run_all() runs all registered graphs (useful when multiple independent pipelines are defined in the same process).

Q: Can I use Pathway in Jupyter notebooks? Yes. Call pw.enable_interactive_mode() at the start. This lets you inspect tables with .show() and iterate interactively without blocking on pw.run().

Q: What data formats does Pathway support? JSON, CSV, plaintext, binary (for documents/PDFs), Avro, and connector-specific formats. Each pw.io.* module documents its supported formats.

Q: How does Pathway handle late-arriving data? The differential dataflow engine processes retractions and corrections automatically. When late data arrives, downstream computations update to reflect the new state — no manual watermarking required for correctness.

Q: Can I test my pipeline without live data? Yes. Use pw.demo.range_stream() or pw.demo.noisy_linear_stream() for synthetic data. Or use pathway record + pathway replay to capture and replay real data.

Q: What happens if my pipeline crashes mid-processing? Without persistence: state is lost, restart from scratch. With persistence enabled (PATHWAY_PERSISTENCE_MODE), the engine records snapshots and resumes from the last consistent state.

Q: Does Pathway support exactly-once semantics? With persistence enabled and idempotent sinks, yes. The engine's snapshot mechanism ensures no data is processed twice after recovery.

Connectors

Q: Which message brokers does Pathway support? Kafka, Redpanda, NATS, MQTT, RabbitMQ, Google Pub/Sub, AWS Kinesis, and Airbyte sources.

Q: Which databases can Pathway write to? PostgreSQL, MongoDB, MySQL, MSSQL, Elasticsearch, SQLite, QuestDB, BigQuery, DynamoDB, Milvus, and more via the pw.io.* modules.

Q: Can I use Pathway with cloud object storage? Yes. pw.io.s3 works with AWS S3, and pw.io.minio works with any S3-compatible storage including MinIO and GCS (via S3 compatibility).

Q: How do I read from Google Drive? Use pw.io.gdrive.read() with appropriate credentials. It monitors a Drive folder for new/updated files.

Q: Can I build a custom connector? Yes. Subclass pw.io.python.ConnectorSubject for sources and pw.io.python.ConnectorObserver for sinks. See the Custom Python Connector recipes above.

Q: Does the Kafka connector support Avro / Schema Registry? Check the pw.io.kafka.read() format parameter and schema registry settings in the connector documentation.

Q: Can I read from multiple Kafka topics? Create separate pw.io.kafka.read() calls for each topic, then combine with pw.Table.concat_reindex().

Q: How do I handle CDC (Change Data Capture) from databases? Use pw.io.debezium which integrates with Debezium CDC connectors, or use the native PostgreSQL/MySQL connectors in CDC mode.

Transformations

Q: Can I use pandas/numpy inside UDFs? Yes. UDFs are regular Python functions — import and use any library inside them.

Q: How do async UDFs handle backpressure? The engine manages concurrency. Configure the executor's capacity to limit outstanding async calls:

@pw.udf(executor=pw.udfs.async_executor(capacity=10))
async def rate_limited_call(x: str) -> str:
    ...

Q: Can I call external APIs from a pipeline? Yes, via async UDFs. Wrap your HTTP calls in an async UDF with appropriate error handling and rate limiting.

Q: How do I deduplicate a stream? Group by the deduplication key and use pw.reducers.latest or pw.reducers.earliest:

deduped = table.groupby(pw.this.unique_key).reduce(
    unique_key=pw.this.unique_key,
    value=pw.reducers.latest(pw.this.value),
)

Q: What temporal operations are available? Tumbling windows, sliding windows, session windows, as-of joins, interval joins, window joins, and inactivity detection. See the Common Tasks section for examples of each.

Q: Can I do multi-way joins? Yes. Chain joins:

result = a.join(b, ...).select(...).join(c, ...).select(...)

Q: How do I handle NULL/None values? Use pw.coalesce() or filter them out with .filter(pw.this.col.is_not_none()).

Scaling and Performance

Q: How many events per second can Pathway handle? Depends on transformation complexity and hardware. Simple filter+select pipelines handle millions of events/sec on a single machine. Complex joins with large state are bounded by memory.

Q: When should I use multiple threads vs. multiple processes? Threads share memory (lower overhead, limited by GIL for Python UDFs but not for engine operations). Processes provide full parallelism but require network communication between workers.

Q: How do I monitor pipeline performance? Set PATHWAY_MONITORING_HTTP_PORT and use pathway web_dashboard or query the metrics endpoint directly.

Q: Can I deploy Pathway on Kubernetes? Yes. Package your pipeline as a Docker container and deploy as a Deployment or StatefulSet. Use persistent volumes for state/persistence directories.

LLM / RAG

Q: What embedding models does VectorStoreServer support? Any model wrapped by the embedder interface. OpenAIEmbedder supports all OpenAI embedding models. Custom embedders can wrap any provider.

Q: Does the vector store update automatically when documents change? Yes. VectorStoreServer watches its input table reactively. When source documents are added, modified, or deleted, the index updates in real time.

Q: How do I parse PDFs and other document types? Use pathway.xpacks.llm.parsers.UnstructuredParser which handles PDFs, DOCX, HTML, and more via the unstructured library.

Q: Can I customize chunk sizes for the splitter? Yes. TokenCountSplitter accepts min_tokens and max_tokens parameters to control chunk boundaries.

Q: How do I query the VectorStoreServer programmatically? POST to /v1/retrieve with a JSON body containing query (string) and k (number of results). The response contains matched chunks with metadata and scores.

CLI

Q: What's the difference between pathway spawn and running python script.py directly? pathway spawn configures multi-threading/multi-process execution, sets up inter-worker communication, and handles recording. For single-threaded execution, they're equivalent.

Q: How do I record and replay data for testing?

# Record
pathway spawn --record --record-path ./recordings my_pipeline.py

# Replay
pathway replay --record-path ./recordings

Q: What does pathway spawn_from_env do? It reads spawn arguments from the PATHWAY_SPAWN_ARGS environment variable. Useful in containerized deployments where command-line arguments are set via environment.

Q: Can I pass arguments to my pipeline script via pathway spawn? Yes. Arguments after the program name are forwarded:

pathway spawn -t 4 my_pipeline.py --config prod.yaml --verbose

Warning

Do not confuse pathway spawn flags (before the program name) with your script's arguments (after the program name). Flags like -t and -n must come before the script path.

CLI Reference

pathway spawn

Run a Pathway program with configurable parallelism.

pathway spawn [OPTIONS] PROGRAM [ARGUMENTS...]

Examples:

# 4 threads, single process
pathway spawn -t 4 my_pipeline.py

# 2 processes, 2 threads each
pathway spawn -n 2 -t 2 my_pipeline.py

# Record all input data
pathway spawn --record --record-path ./test_data my_pipeline.py

# Multi-machine (2 machines, 2 processes each)
pathway spawn -n 2 --addresses "host1:10000,host1:10001,host2:10000,host2:10001" -pi 0 pipeline.py

pathway replay

Replay previously recorded connector data.

pathway replay [OPTIONS]

pathway spawn_from_env

Spawn using arguments from the PATHWAY_SPAWN_ARGS environment variable.

export PATHWAY_SPAWN_ARGS="-t 4 my_pipeline.py --config prod.yaml"
pathway spawn_from_env

pathway web_dashboard

Launch the monitoring web dashboard.

pathway web_dashboard [OPTIONS]

pathway web_dashboard --port 9090 --detailed-metrics-dir ./metrics

pathway airbyte create_source

Create an Airbyte-compatible source connector.

pathway airbyte create_source [OPTIONS] CONNECTION

Connector Quick Reference

Caution

Connector availability may depend on optional dependencies. If importing a connector raises an ImportError, install the relevant extra (e.g., pip install "pathway[kafka]" or check the specific connector's documentation for required packages).

Best Practices

Pipeline Design

1. Define all inputs and outputs before pw.run() — The computation graph is built lazily; pw.run() executes it.

2. Use schemas — Explicit schemas catch type errors at graph construction time rather than at runtime.

3. Prefer built-in operations over UDFs — Pathway's engine optimizes native operations (filter, select, join) much better than Python UDFs.

4. Bound your windows — Unbounded temporal windows accumulate state indefinitely. Always specify a duration.

5. Test with batch, deploy with streaming — Write pipelines against static files first. The same code handles live streams without changes.

Production Deployment

1. Enable persistence for stateful pipelines that must survive restarts.

2. Monitor with PATHWAY_MONITORING_HTTP_PORT — Always set this in production to observe throughput, latency, and memory usage.

3. Use pathway spawn in production for proper process management, rather than raw python script.py.

4. Record inputs in staging with --record to create reproducible test fixtures for debugging production issues.

5. Handle UDF errors gracefully — A single UDF exception can crash the entire pipeline. Always include error handling in UDFs.

Tip

For architecture details, persistence backend configuration, security considerations, multi-node deployment topology, and engine internals, see the Admin Guide at Admin Guide.