Pathway Tutorial: Real-Time Data Processing Framework for AI and Machine Learning

Pathway is an open-source Python framework designed specifically for building real-time data pipelines. Unlike traditional batch processing tools, Pathway enables you to process streaming data with a pandas-like syntax, making it an excellent choice for building AI applications that require up-to-date data such as Retrieval-Augmented Generation (RAG), real-time monitoring, and feature engineering.

In this tutorial, we will learn how to use Pathway from installation to building production-ready real-time pipelines, including LLM integration for building RAG systems that stay automatically updated.

Why Pathway?

Before diving into implementation, it is important to understand why Pathway stands out in the data engineering ecosystem:

Unified Batch and Streaming: One codebase for both batch and streaming processing. You write code once and can run it in both modes.

Pandas-like Syntax: If you are already familiar with pandas, transitioning to Pathway feels natural. Operations like select, filter, groupby, and join are available with a consistent API.

Incremental Computation: Pathway only recomputes the parts that changed, not the entire dataset. This makes processing very efficient for continuously changing data.

Built-in Connectors: Supports various data sources including Kafka, PostgreSQL, file system, Google Drive, SharePoint, and many more.

LLM Integration: Pathway has built-in modules for building real-time RAG pipelines with support for OpenAI, LiteLLM, and other embedding models.

Installation

Basic Installation

Pathway can be installed using pip. Make sure you are using Python 3.10 or later:

pip install pathway

Installation with LLM Features

If you plan to use Pathway for RAG pipelines or LLM integration:

pip install "pathway[xpack-llm]"

Installation with All Features

To get all features including additional connectors:

pip install "pathway[all]"

Verify Installation

import pathway as pw
print(pw.version)

Setup Environment Variables

For LLM features, you need to set up your API key:

export OPENAIAPIKEY="sk-your-api-key-here"

Core Concepts

Tables

The core concept in Pathway is the Table. Similar to a DataFrame in pandas, a Table is a tabular data structure that holds your data. The key difference is that Tables in Pathway are reactive to data changes.

import pathway as pw

Create a table from static data (for testing)
data = pw.debug.tablefrommarkdown("""
    | name      | department  | salary
1   | Alice     | Engineering | 95000
2   | Bob       | Marketing   | 78000
3   | Carol     | Engineering | 92000
4   | David     | Marketing   | 82000
5   | Eve       | Engineering | 98000
""")

Display schema
print(data.schema)

Transformations

Pathway supports various familiar transformation operations:

import pathway as pw

data = pw.debug.tablefrommarkdown("""
    | name      | department  | salary
1   | Alice     | Engineering | 95000
2   | Bob       | Marketing   | 78000
3   | Carol     | Engineering | 92000
4   | David     | Marketing   | 82000
5   | Eve       | Engineering | 98000
""")

Select specific columns
selected = data.select(data.name, data.salary)

Filter data
engineers = data.filter(data.department == "Engineering")

Add new columns
withbonus = data.select(

    pw.this,

    bonus=pw.this.salary  0.1
)

GroupBy and aggregation
deptstats = data.groupby(data.department).reduce(
    data.department,
    totalsalary=pw.reducers.sum(data.salary),

    employeecount=pw.reducers.count(),
    avgsalary=pw.reducers.avg(data.salary)

)

pw.debug.computeandprint(deptstats)

Connectors

Pathway provides various connectors for reading and writing data:

import pathway as pw

Read from CSV files (will monitor for changes)
table = pw.io.csv.read(
    "./data/",
    schema=pw.schemafromdict({
        "name": str,
        "value": float,
        "timestamp": str
    }),
    mode="streaming"
)

Read from Kafka
kafkatable = pw.io.kafka.read(

    rdkafkasettings={
        "bootstrap.servers": "localhost:9092",
        "group.id": "pathway-consumer"
    },
    topic="sensor-data",
    format="json",
    schema=pw.schemafromdict({
        "sensorid": str,

        "value": float,
        "timestamp": int
    })
)

Write to JSON Lines file
pw.io.jsonlines.write(table, "./output/results.jsonl")

Write to PostgreSQL
pw.io.postgres.write(
    table,
    connectionstring="postgresql://user:password@localhost:5432/mydb",
    tablename="results"

)

Building a Real-Time ETL Pipeline

Let us build a real-time ETL pipeline that processes IoT sensor data:

import pathway as pw
from datetime import datetime

Schema for sensor data
class SensorSchema(pw.Schema):
    sensorid: str
    temperature: float
    humidity: float
    timestamp: int

Read data from a folder (Pathway will monitor for new files)
rawdata = pw.io.csv.read(

    "./sensordata/",
    schema=SensorSchema,
    mode="streaming"
)

Transform: convert temperature and add columns
processed = rawdata.select(

    pw.this,

    tempcelsius=pw.this.temperature,
    tempfahrenheit=pw.this.temperature  9 / 5 + 32,
    ishightemp=pw.this.temperature > 35.0,

    ishighhumidity=pw.this.humidity > 80.0

)

Filter: only abnormal conditions
alerts = processed.filter(
    (pw.this.ishightemp) | (pw.this.ishighhumidity)

)

Aggregate per sensor: average temperature and humidity
sensorstats = processed.groupby(pw.this.sensorid).reduce(

    pw.this.sensorid,
    avgtemperature=pw.reducers.avg(pw.this.tempcelsius),
    maxtemperature=pw.reducers.max(pw.this.tempcelsius),
    avghumidity=pw.reducers.avg(pw.this.humidity),

    readingcount=pw.reducers.count()
)

Output: save results to files
pw.io.jsonlines.write(alerts, "./output/alerts.jsonl")
pw.io.jsonlines.write(sensorstats, "./output/sensorstats.jsonl")

Run the pipeline
pw.run()

This pipeline runs continuously, monitoring the sensordata/ folder for new files, and automatically updates all transformations and outputs when new data arrives.

Temporal Joins and Windowing

Pathway supports temporal operations useful for time-series data:

import pathway as pw

Transaction data
transactions = pw.debug.tablefrommarkdown("""

    | userid | amount | t
1   | usera  | 100    | 1

2   | userb  | 250    | 2
3   | usera  | 75     | 3

4   | userc  | 500    | 4
5   | usera  | 200    | 5

6   | userb  | 150    | 6
""")

Exchange rate data
exchangerates = pw.debug.tablefrommarkdown("""

    | currency | rate  | t
1   | EUR      | 1.08  | 1
2   | EUR      | 1.09  | 3
3   | EUR      | 1.07  | 5
""")

Temporal join: combine transactions with latest exchange rate
result = transactions.asofjoin(
    exchangerates,

    transactions.t,
    exchangerates.t,
    how="left"
).select(
    pw.left.userid,

    pw.left.amount,
    pw.left.t,
    exchangerate=pw.right.rate
)

pw.debug.computeandprint(result)

Sliding Window Aggregation

import pathway as pw

sensordata = pw.debug.tablefrommarkdown("""

    | sensorid | value | timestamp
1   | s1        | 25.0  | 1000
2   | s1        | 26.5  | 2000
3   | s1        | 24.0  | 3000
4   | s1        | 27.0  | 4000
5   | s1        | 28.5  | 5000
6   | s2        | 30.0  | 1000
7   | s2        | 31.0  | 2000
8   | s2        | 29.5  | 3000
""")

Sliding window: average per 3 seconds with 1 second slide
windowed = sensordata.windowby(

    sensordata.timestamp,
    window=pw.temporal.sliding(
        duration=3000,
        hop=1000
    ),
    instance=sensordata.sensorid
).reduce(
    sensorid=pw.this.pwinstance,

    windowstart=pw.this.pwwindowstart,

    windowend=pw.this.pwwindowend,

    avgvalue=pw.reducers.avg(pw.this.value),
    maxvalue=pw.reducers.max(pw.this.value),

    count=pw.reducers.count()
)

pw.debug.computeandprint(windowed)

Real-Time RAG with Pathway

One of Pathway's standout features is its ability to build RAG pipelines that automatically update the knowledge base when documents change. This is extremely useful for enterprise chatbots that need to always have the latest information.

Setting Up a RAG Pipeline

import pathway as pw
from pathway.xpacks.llm.documentstore import DocumentStore
from pathway.xpacks.llm.servers import DocumentStoreServer
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.parsers import UnstructuredParser

Configuration: monitor document folder
datasources = []

datasources.append(
    pw.io.fs.read(
        path="./documents/",
        format="binary",
        mode="streaming",
        withmetadata=True

    )
)

Setup text splitter
splitter = TokenCountSplitter(
    mintokens=100,
    maxtokens=500,

    encodingname="cl100kbase"

)

Setup embedder
embedder = OpenAIEmbedder(
    model="text-embedding-3-small",
    apikey=pw.secrets.load("OPENAIAPIKEY")
)

Setup parser for various document formats
parser = UnstructuredParser()

Create document store
documentstore = DocumentStore(

    docs=datasources,
    splitter=splitter,
    embedder=embedder,
    parser=parser
)

Run REST API server
server = DocumentStoreServer(
    host="0.0.0.0",
    port=8765,
    documentstore=documentstore
)
server.run()

Querying the RAG from a Client

Once the server is running, you can query through the REST API:

import requests

Query the document store
response = requests.post(
    "http://localhost:8765/v1/retrieve",
    json={
        "query": "How to optimize data pipelines?",
        "k": 5
    }
)

results = response.json()
for doc in results:
    print(f"Score: {doc['dist']:.4f}")
    print(f"Content: {doc['text'][:200]}...")
    print("---")

Complete RAG Pipeline with LLM

import pathway as pw
from pathway.xpacks.llm.questionanswering import (

    SummaryQuestionAnswerer,
)
from pathway.xpacks.llm.documentstore import DocumentStore
from pathway.xpacks.llm.servers import QASummaryRestServer
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.llms import OpenAIChat

Data source: document folder
docs = pw.io.fs.read(
    path="./knowledgebase/",

    format="binary",
    mode="streaming",
    withmetadata=True
)

Configure components
embedder = OpenAIEmbedder(model="text-embedding-3-small")
splitter = TokenCountSplitter(maxtokens=400)

llm = OpenAIChat(
    model="gpt-4o-mini",
    temperature=0.0
)

Create document store
docstore = DocumentStore(
    docs=[docs],
    splitter=splitter,
    embedder=embedder
)

Setup QA pipeline
qa = SummaryQuestionAnswerer(
    llm=llm,
    documentstore=docstore
)

Run server
server = QASummaryRestServer(
    host="0.0.0.0",
    port=8765,
    questionanswerer=qa

)
server.run()

With this setup, every time you add, modify, or delete files in the knowledgebase/ folder, the index is automatically updated without restarting the server.

Monitoring and Real-Time Alerting

Pathway can be used to build responsive monitoring systems:

import pathway as pw

Schema for application logs
class LogSchema(pw.Schema):
    timestamp: int
    level: str
    service: str
    message: str
    responsetimems: float

Read logs from Kafka
logs = pw.io.kafka.read(
    rdkafkasettings={

        "bootstrap.servers": "localhost:9092",
        "group.id": "log-monitor"
    },
    topic="application-logs",
    format="json",
    schema=LogSchema
)

Calculate error rate per service in 5-minute windows
errorcounts = logs.filter(
    pw.this.level == "ERROR"
).windowby(
    pw.this.timestamp,
    window=pw.temporal.tumbling(duration=300000),
    instance=pw.this.service
).reduce(
    service=pw.this.pwinstance,
    windowend=pw.this.pwwindowend,
    errorcount=pw.reducers.count()

)

Calculate average response time per service
responsestats = logs.windowby(
    pw.this.timestamp,
    window=pw.temporal.tumbling(duration=300000),
    instance=pw.this.service
).reduce(
    service=pw.this.pwinstance,
    windowend=pw.this.pwwindowend,
    avgresponsetime=pw.reducers.avg(pw.this.responsetimems),
    p99responsetime=pw.reducers.max(pw.this.responsetimems),
    requestcount=pw.reducers.count()

)

Alert: services with more than 10 errors in 5 minutes
higherroralerts = errorcounts.filter(pw.this.errorcount > 10)


Output alerts
pw.io.jsonlines.write(higherroralerts, "./alerts/erroralerts.jsonl")
pw.io.jsonlines.write(responsestats, "./metrics/responsestats.jsonl")

pw.run()

Integration with Google Drive and SharePoint

Pathway supports monitoring documents from cloud storage:

import pathway as pw
from pathway.xpacks.llm.documentstore import DocumentStore

from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.embedders import OpenAIEmbedder

Read from Google Drive
gdrivedocs = pw.io.gdrive.read(
    objectid="your-folder-id",

    serviceusercredentialsfile="credentials.json",
    mode="streaming",
    withmetadata=True,

    refreshinterval=30
)

Also read from local filesystem
localdocs = pw.io.fs.read(

    path="./localdocs/",
    format="binary",
    mode="streaming",
    withmetadata=True

)

Combine all sources
alldocs = gdrivedocs + localdocs

Create document store with all sources
embedder = OpenAIEmbedder(model="text-embedding-3-small")
splitter = TokenCountSplitter(maxtokens=400)


docstore = DocumentStore(
    docs=[alldocs],

    splitter=splitter,
    embedder=embedder
)

Best Practices

1. Schema Definition

Always define schemas explicitly for better type checking and error messages:

import pathway as pw

class TransactionSchema(pw.Schema):
    transactionid: str
    userid: str

    amount: float
    currency: str
    timestamp: int
    category: str

transactions = pw.io.kafka.read(
    rdkafkasettings={"bootstrap.servers": "localhost:9092"},
    topic="transactions",
    format="json",
    schema=TransactionSchema
)

2. Error Handling in Pipelines

import pathway as pw

rawdata = pw.io.csv.read("./data/", schema=MySchema, mode="streaming")


Use pw.coalesce to handle null values
cleaned = rawdata.select(
    pw.this,

    amount=pw.coalesce(pw.this.amount, 0.0),
    category=pw.coalesce(pw.this.category, "unknown")
)

Filter invalid data
validdata = cleaned.filter(

    (pw.this.amount >= 0) & (pw.this.category != "unknown")
)

3. Performance Optimization

import pathway as pw Use persistence for state checkpointing pw.run(monitoringlevel=pw.MonitoringLevel.ALL, persistenceconfig=pw.persistence.Config( pw.persistence.Backend.filesystem("./checkpoints/"), snapshotinterval=300 ) )

4. Testing Pipelines

import pathway as pw

def testtransformation():

    inputdata = pw.debug.tablefrommarkdown("""
        | name  | score
    1   | Alice | 85
    2   | Bob   | 92
    3   | Carol | 78
    """)

    result = inputdata.select(

        pw.this,
        passed=pw.this.score >= 80
    )

    expected = pw.debug.tablefrommarkdown("""

        | name  | score | passed
    1   | Alice | 85    | True
    2   | Bob   | 92    | True
    3   | Carol | 78    | False
    """)

    pw.debug.computeandprint(result)

    # Verify output matches expectations

5. Deployment with Docker

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "pipeline.py"]

# docker-compose.yml
services:
  pathway-pipeline:
    build: .
    volumes:
./data:/app/data
./output:/app/output
./checkpoints:/app/checkpoints
    environment:
OPENAIAPIKEY=${OPENAIAPIKEY}

    restart: unless-stopped

  kafka:
    image: confluentinc/cp-kafka:latest
    ports:
"9092:9092"
    environment:
      KAFKAADVERTISEDLISTENERS: PLAINTEXT://localhost:9092

Full Project Example: Real-Time Customer Support Knowledge Base

Here is a complete example of building a knowledge base for customer support that updates automatically:

import pathway as pw
from pathway.xpacks.llm.documentstore import DocumentStore
from pathway.xpacks.llm.servers import QASummaryRestServer
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.llms import OpenAIChat
from pathway.xpacks.llm.questionanswering import SummaryQuestionAnswerer


Document sources: FAQ, policy docs, product docs
faqdocs = pw.io.fs.read(
    path="./docs/faq/",
    format="binary",
    mode="streaming",
    withmetadata=True

)

policydocs = pw.io.fs.read(
    path="./docs/policies/",
    format="binary",
    mode="streaming",
    withmetadata=True

)

productdocs = pw.io.fs.read(
    path="./docs/products/",
    format="binary",
    mode="streaming",
    withmetadata=True

)

Combine all documents
alldocs = faqdocs + policydocs + productdocs


Setup components
embedder = OpenAIEmbedder(model="text-embedding-3-small")
splitter = TokenCountSplitter(mintokens=100, maxtokens=500)


llm = OpenAIChat(
    model="gpt-4o-mini",
    temperature=0.0
)

Document store
docstore = DocumentStore(
    docs=[alldocs],

    splitter=splitter,
    embedder=embedder
)

QA system
qa = SummaryQuestionAnswerer(
    llm=llm,
    documentstore=docstore

)

REST API server
server = QASummaryRestServer(
    host="0.0.0.0",
    port=8080,
    questionanswerer=qa
)

print("Starting Customer Support Knowledge Base Server...")
print("API available at http://localhost:8080")
print("Monitoring folders: docs/faq/, docs/policies/, docs/products/")
server.run()

Conclusion

Pathway is a highly useful framework for building real-time data pipelines with Python. Its key advantages include:

Unified API for both batch and streaming that makes development more efficient
Incremental computation ensuring only changes are reprocessed
Built-in RAG support that simplifies creating knowledge bases that stay up-to-date
Rich connector ecosystem for various data sources
Great developer experience with pandas-like syntax

Pathway is well suited for use cases such as:

Real-time RAG and enterprise chatbots
Monitoring and alerting systems
Streaming ETL pipelines
Real-time feature engineering for ML
Document processing pipelines

To get started, simply install Pathway with pip install pathway and follow the examples above. Comprehensive documentation is available on the official Pathway website for exploring more advanced features.