# 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

```bash
pip install -U pathway
```

```python
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:

```bash
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

| Requirement | Version |
|---|---|
| Python | 3.10 or higher |
| OS | Linux, macOS, Windows (via WSL2) |
| Architecture | x86_64, aarch64 |

### Standard Install

```bash
pip install -U pathway
```

### With Optional Dependencies

```bash
# 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

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

### Docker

```bash
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:

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

Or via environment variable:

```bash
export PATHWAY_LICENSE_KEY="your-key-here"
```

### Environment Variables

| Variable | Purpose | Default |
|---|---|---|
| `PATHWAY_LICENSE_KEY` | License key for Scale features | (none) |
| `PATHWAY_PERSISTENCE_MODE` | Persistence mode for state recovery | (none) |
| `PATHWAY_REPLAY_MODE` | Replay mode for recorded data | (none) |
| `PATHWAY_SNAPSHOT_ACCESS` | Snapshot access mode (`record` or `replay`) | (none) |
| `PATHWAY_SPAWN_ARGS` | Arguments for `spawn_from_env` CLI command | (none) |
| `PATHWAY_MONITORING_HTTP_PORT` | Port for the HTTP monitoring endpoint | (none) |
| `PATHWAY_THREADS` | Number of threads (set by `spawn`) | 1 |
| `PATHWAY_PROCESSES` | Number of processes (set by `spawn`) | 1 |
| `PATHWAY_SKIP_START_LOG` | Skip startup log banner | (none) |

### YAML Pipeline Configuration

Pathway supports defining entire pipelines declaratively in YAML:

```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:

```python
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.

```python
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:

```python
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:

| Reference | Usage |
|---|---|
| `pw.this` | Current table columns |
| `pw.left` | Left table in a join |
| `pw.right` | Right table in a join |

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

### Filtering

```python
# 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

```python
# 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

```python
# 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

```python
# 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

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

### SQL Interface

```python
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)

```python
# 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))
```

```python
# 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

```python
# 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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:

```python
# 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:

```python
# 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

```python
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:

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

### Task: Multi-Source Merge

```python
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

```python
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:

```bash
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:

```bash
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)

```python
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)

```python
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

```python
# 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

```python
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

```python
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)

```python
# 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

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

### Task: Delta Lake / Iceberg Output

```python
# 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

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

### Task: Add Update Timestamps

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

### Task: Demo / Testing Streams

```python
# 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:
```bash
# 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:
```bash
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:
```python
@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:
```bash
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:
```bash
pathway spawn -n 4 --first-port 11000 my_pipeline.py
```

**Problem: Multi-machine deployment not distributing work**
Use `--addresses` to specify all nodes:
```bash
# 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:
```bash
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()`:
```bash
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:
```python
@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`:
```python
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:
```python
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?**
```bash
# 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:
```bash
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.

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

| Flag | Default | Description |
|---|---|---|
| `-t, --threads` | 1 | Threads per process |
| `-n, --processes` | 1 | Number of processes |
| `--first-port` | 10000 | First port for inter-process communication |
| `--addresses` | (auto) | Comma-separated host:port for multi-machine |
| `-pi, --process-id` | (auto) | Process index on current machine (required with --addresses) |
| `--record` | off | Record input connector data |
| `--record-path` | "record" | Directory for recordings |
| `--repository-url` | (none) | GitHub repo path if spawned from a repo |
| `--branch` | (none) | Branch if different from default |

Examples:

```bash
# 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.

```text
pathway replay [OPTIONS]
```

| Flag | Default | Description |
|---|---|---|
| `-t, --threads` | 1 | Threads per process |
| `-n, --processes` | 1 | Number of processes |
| `--first-port` | 10000 | First port for inter-process communication |
| `--record-path` | "record" | Directory where recording is stored |
| `--mode` | (default) | Mode of replaying data |
| `--continue` | off | Continue with realtime data after replay |
| `--repository-url` | (none) | GitHub repo path |
| `--branch` | (none) | Branch |

### `pathway spawn_from_env`

Spawn using arguments from the `PATHWAY_SPAWN_ARGS` environment variable.

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

### `pathway web_dashboard`

Launch the monitoring web dashboard.

```text
pathway web_dashboard [OPTIONS]
```

| Flag | Default | Description |
|---|---|---|
| `--detailed-metrics-dir` | "." | Directory containing metrics data |
| `--port` | 8088 | Dashboard port |

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

### `pathway airbyte create_source`

Create an Airbyte-compatible source connector.

```text
pathway airbyte create_source [OPTIONS] CONNECTION
```

| Flag | Default | Description |
|---|---|---|
| `--image` | "airbyte/source-faker:0.1.4" | Docker image for the Airbyte source |

---

## Connector Quick Reference

| Module | Read | Write | Protocol/Format |
|---|---|---|---|
| `pw.io.kafka` | Yes | Yes | Apache Kafka |
| `pw.io.redpanda` | Yes | Yes | Redpanda (Kafka-compatible) |
| `pw.io.s3` | Yes | Yes | AWS S3 |
| `pw.io.minio` | Yes | Yes | S3-compatible (MinIO) |
| `pw.io.fs` | Yes | — | Local filesystem (streaming) |
| `pw.io.csv` | Yes | Yes | CSV files |
| `pw.io.jsonlines` | Yes | Yes | JSON Lines files |
| `pw.io.plaintext` | Yes | — | Plain text files |
| `pw.io.http` | Yes | — | HTTP server (webhooks) |
| `pw.io.postgres` | Yes | Yes | PostgreSQL |
| `pw.io.mongodb` | Yes | Yes | MongoDB |
| `pw.io.mysql` | Yes | Yes | MySQL |
| `pw.io.mssql` | Yes | Yes | Microsoft SQL Server |
| `pw.io.sqlite` | Yes | Yes | SQLite |
| `pw.io.elasticsearch` | — | Yes | Elasticsearch |
| `pw.io.dynamodb` | Yes | Yes | AWS DynamoDB |
| `pw.io.bigquery` | Yes | Yes | Google BigQuery |
| `pw.io.kinesis` | Yes | Yes | AWS Kinesis |
| `pw.io.nats` | Yes | Yes | NATS |
| `pw.io.mqtt` | Yes | Yes | MQTT |
| `pw.io.rabbitmq` | Yes | Yes | RabbitMQ |
| `pw.io.pubsub` | Yes | Yes | Google Pub/Sub |
| `pw.io.gdrive` | Yes | — | Google Drive |
| `pw.io.slack` | Yes | — | Slack |
| `pw.io.debezium` | Yes | — | Debezium CDC |
| `pw.io.airbyte` | Yes | — | Airbyte connectors |
| `pw.io.iceberg` | — | Yes | Apache Iceberg |
| `pw.io.deltalake` | — | Yes | Delta Lake |
| `pw.io.questdb` | — | Yes | QuestDB |
| `pw.io.milvus` | — | Yes | Milvus vector DB |
| `pw.io.python` | Yes | Yes | Custom Python |
| `pw.io.null` | — | Yes | Discard (dev/testing) |

> [!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](/pathwaycom/pathway/admin-guide/).

---

_Generated May 2026 from pathwaycom/pathway at f0afd65. AI-generated documentation. The library is checked by sampling (three repositories so far); this guide was not individually reviewed. Verify against the project before relying on it. Unofficial; not affiliated with or reviewed by the project._