Tutorial Pathway: Framework Real-Time Data Processing untuk AI dan Machine Learning

Pathway adalah framework open-source Python yang dirancang khusus untuk membangun pipeline data real-time. Berbeda dari tools batch processing tradisional, Pathway memungkinkan Anda memproses data secara streaming dengan sintaks yang mirip pandas, menjadikannya pilihan yang tepat untuk membangun aplikasi AI yang membutuhkan data terkini seperti Retrieval-Augmented Generation (RAG), monitoring real-time, dan feature engineering.

Dalam tutorial ini, kita akan mempelajari cara menggunakan Pathway dari instalasi hingga membangun pipeline real-time yang siap produksi, termasuk integrasi dengan LLM untuk membangun sistem RAG yang selalu ter-update.

Mengapa Pathway?

Sebelum masuk ke implementasi, penting untuk memahami mengapa Pathway menjadi pilihan yang menarik di ekosistem data engineering:

Unified Batch dan Streaming: Satu codebase untuk batch dan streaming processing. Anda menulis kode sekali dan bisa menjalankannya untuk kedua mode.

Sintaks Mirip Pandas: Jika Anda sudah familiar dengan pandas, transisi ke Pathway akan terasa natural. Operasi seperti select, filter, groupby, dan join tersedia dengan API yang konsisten.

Incremental Computation: Pathway hanya menghitung ulang bagian yang berubah, bukan seluruh dataset. Ini membuat pemrosesan sangat efisien untuk data yang terus berubah.

Built-in Connectors: Mendukung berbagai sumber data seperti Kafka, PostgreSQL, file system, Google Drive, SharePoint, dan banyak lagi.

LLM Integration: Pathway memiliki modul bawaan untuk membangun pipeline RAG real-time dengan dukungan untuk OpenAI, LiteLLM, dan embedding models lainnya.

Instalasi

Instalasi Dasar

Pathway bisa diinstal menggunakan pip. Pastikan Anda menggunakan Python 3.10 atau lebih baru:

pip install pathway

Instalasi dengan Fitur LLM

Jika Anda berencana menggunakan Pathway untuk pipeline RAG atau integrasi LLM:

pip install "pathway[xpack-llm]"

Instalasi dengan Semua Fitur

Untuk mendapatkan semua fitur termasuk connectors tambahan:

pip install "pathway[all]"

Verifikasi Instalasi

import pathway as pw
print(pw.version)

Setup Environment Variables

Untuk fitur LLM, Anda perlu mengatur API key:

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

Konsep Dasar Pathway

Tables

Konsep utama dalam Pathway adalah Table. Mirip dengan DataFrame di pandas, Table adalah struktur data tabular yang menyimpan data Anda. Perbedaan utamanya adalah Table di Pathway bersifat reaktif terhadap perubahan data.

import pathway as pw

Membuat table dari data statis (untuk testing)
data = pw.debug.tablefrommarkdown("""
    | nama      | departemen | gaji
1   | Andi      | Engineering | 15000000
2   | Budi      | Marketing   | 12000000
3   | Citra     | Engineering | 14000000
4   | Dewi      | Marketing   | 13000000
5   | Eko       | Engineering | 16000000
""")

Menampilkan schema
print(data.schema)

Transformasi

Pathway mendukung berbagai operasi transformasi yang familiar:

import pathway as pw

data = pw.debug.tablefrommarkdown("""
    | nama      | departemen  | gaji
1   | Andi      | Engineering | 15000000
2   | Budi      | Marketing   | 12000000
3   | Citra     | Engineering | 14000000
4   | Dewi      | Marketing   | 13000000
5   | Eko       | Engineering | 16000000
""")

Select kolom tertentu
selected = data.select(data.nama, data.gaji)

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

Menambah kolom baru
withbonus = data.select(

    pw.this,

    bonus=pw.this.gaji  0.1
)

GroupBy dan agregasi
deptstats = data.groupby(data.departemen).reduce(
    data.departemen,
    totalgaji=pw.reducers.sum(data.gaji),

    jumlahkaryawan=pw.reducers.count(),
    rataratagaji=pw.reducers.avg(data.gaji)
)

pw.debug.computeandprint(deptstats)

Connectors

Pathway menyediakan berbagai connector untuk membaca dan menulis data:

import pathway as pw

Membaca dari file CSV (akan memantau perubahan file)
table = pw.io.csv.read(
    "./data/",
    schema=pw.schemafromdict({

        "nama": str,
        "nilai": float,
        "timestamp": str
    }),
    mode="streaming"
)

Membaca dari 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
    })
)

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

Menulis ke PostgreSQL
pw.io.postgres.write(
    table,
    connectionstring="postgresql://user:password@localhost:5432/mydb",

    tablename="results"
)

Membangun Pipeline ETL Real-Time

Mari kita bangun pipeline ETL real-time yang memproses data sensor IoT:

import pathway as pw
from datetime import datetime

Schema untuk data sensor
class SensorSchema(pw.Schema):
    sensorid: str

    temperature: float
    humidity: float
    timestamp: int

Membaca data dari folder (Pathway akan memantau file baru)
rawdata = pw.io.csv.read(
    "./sensordata/",

    schema=SensorSchema,
    mode="streaming"
)

Transformasi: konversi timestamp dan tambah kolom
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: hanya data dengan kondisi abnormal
alerts = processed.filter(
    (pw.this.ishightemp) | (pw.this.ishighhumidity)
)

Agregasi per sensor: rata-rata suhu dan kelembapan
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: simpan hasil ke file
pw.io.jsonlines.write(alerts, "./output/alerts.jsonl")
pw.io.jsonlines.write(sensorstats, "./output/sensorstats.jsonl")


Jalankan pipeline
pw.run()

Pipeline di atas akan berjalan terus-menerus, memantau folder sensordata/ untuk file baru, dan secara otomatis memperbarui semua transformasi dan output ketika data baru masuk.

Temporal Joins dan Windowing

Pathway mendukung operasi temporal yang berguna untuk data time-series:

import pathway as pw

Data transaksi
transactions = pw.debug.tablefrommarkdown("""
    | userid | amount  | t

1   | usera  | 100000  | 1
2   | userb  | 250000  | 2

3   | usera  | 75000   | 3
4   | userc  | 500000  | 4

5   | usera  | 200000  | 5
6   | userb  | 150000  | 6

""")

Data kurs mata uang
exchangerates = pw.debug.tablefrommarkdown("""
    | currency | rate    | t
1   | USD      | 15500   | 1
2   | USD      | 15600   | 3
3   | USD      | 15450   | 5
""")

Temporal join: gabungkan transaksi dengan kurs terbaru
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: rata-rata per 3 detik dengan slide 1 detik
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 dengan Pathway

Salah satu fitur unggulan Pathway adalah kemampuannya membangun pipeline RAG yang secara otomatis memperbarui knowledge base ketika dokumen berubah. Ini sangat berguna untuk chatbot perusahaan yang perlu selalu memiliki informasi terkini.

Setup 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

Konfigurasi: pantau folder dokumen
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 untuk berbagai format dokumen
parser = UnstructuredParser()

Buat document store
documentstore = DocumentStore(
    docs=datasources,

    splitter=splitter,
    embedder=embedder,
    parser=parser
)

Jalankan server REST API
server = DocumentStoreServer(
    host="0.0.0.0",
    port=8765,
    documentstore=documentstore

)
server.run()

Query RAG dari Client

Setelah server berjalan, Anda bisa melakukan query melalui REST API:

import requests Query ke document store response = requests.post( "http://localhost:8765/v1/retrieve", json={ "query": "Bagaimana cara mengoptimalkan pipeline data?", "k": 5 } ) results = response.json() for doc in results: print(f"Score: {doc['dist']:.4f}") print(f"Content: {doc['text'][:200]}...") print("---")

RAG Pipeline Lengkap dengan 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: folder dokumen
docs = pw.io.fs.read(
    path="./knowledgebase/",
    format="binary",
    mode="streaming",
    withmetadata=True

)

Konfigurasi komponen
embedder = OpenAIEmbedder(model="text-embedding-3-small")
splitter = TokenCountSplitter(maxtokens=400)
llm = OpenAIChat(
    model="gpt-4o-mini",
    temperature=0.0
)

Buat document store
docstore = DocumentStore(

    docs=[docs],
    splitter=splitter,
    embedder=embedder
)

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

)

Jalankan server
server = QASummaryRestServer(
    host="0.0.0.0",
    port=8765,
    questionanswerer=qa
)
server.run()

Dengan setup ini, setiap kali Anda menambah, mengubah, atau menghapus file di folder knowledgebase/, index akan otomatis di-update tanpa perlu restart server.

Monitoring dan Alerting Real-Time

Pathway bisa digunakan untuk membangun sistem monitoring yang responsif:

import pathway as pw

Schema untuk log aplikasi
class LogSchema(pw.Schema):
    timestamp: int
    level: str
    service: str
    message: str
    responsetimems: float


Baca log dari Kafka
logs = pw.io.kafka.read(
    rdkafkasettings={
        "bootstrap.servers": "localhost:9092",
        "group.id": "log-monitor"
    },
    topic="application-logs",
    format="json",
    schema=LogSchema
)

Hitung error rate per service dalam window 5 menit
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()
)

Hitung rata-rata 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: service dengan error > 10 dalam 5 menit
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()

Integrasi dengan Google Drive dan SharePoint

Pathway mendukung pemantauan dokumen dari 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

Membaca dari Google Drive
gdrivedocs = pw.io.gdrive.read(

    objectid="your-folder-id",
    serviceusercredentialsfile="credentials.json",

    mode="streaming",
    withmetadata=True,
    refreshinterval=30

)

Membaca dari local filesystem juga
localdocs = pw.io.fs.read(
    path="./localdocs/",

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

Gabungkan semua sumber
alldocs = gdrivedocs + localdocs


Buat document store dengan semua sumber
embedder = OpenAIEmbedder(model="text-embedding-3-small")
splitter = TokenCountSplitter(maxtokens=400)

docstore = DocumentStore(

    docs=[alldocs],
    splitter=splitter,
    embedder=embedder
)

Best Practices

1. Schema Definition

Selalu definisikan schema secara eksplisit untuk mendapatkan type checking dan error messages yang lebih baik:

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 dalam Pipeline

import pathway as pw

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

Gunakan pw.coalesce untuk menangani nilai null
cleaned = rawdata.select(

    pw.this,

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

Filter data yang tidak valid
validdata = cleaned.filter(
    (pw.this.amount >= 0) & (pw.this.category != "unknown")
)

3. Optimasi Performa

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

4. Testing Pipeline

import pathway as pw

def testtransformation():
    inputdata = pw.debug.tablefrommarkdown("""

        | nama  | nilai
    1   | Andi  | 85
    2   | Budi  | 92
    3   | Citra | 78
    """)

    result = inputdata.select(
        pw.this,
        lulus=pw.this.nilai >= 80
    )

    expected = pw.debug.tablefrommarkdown("""
        | nama  | nilai | lulus
    1   | Andi  | 85    | True
    2   | Budi  | 92    | True
    3   | Citra | 78    | False
    """)

    pw.debug.computeandprint(result)
    # Verifikasi output sesuai ekspektasi

5. Deployment dengan 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

Contoh Proyek: Real-Time Customer Support Knowledge Base

Berikut contoh lengkap membangun knowledge base untuk customer support yang otomatis update:

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

Sumber dokumen: 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
)

Gabungkan semua dokumen
alldocs = faqdocs + policydocs + productdocs

Setup komponen
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,
    question_answerer=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()

Kesimpulan

Pathway adalah framework yang sangat berguna untuk membangun pipeline data real-time dengan Python. Keunggulan utamanya meliputi:

Unified API untuk batch dan streaming yang membuat development lebih efisien
Incremental computation yang memastikan hanya perubahan yang diproses ulang
Built-in RAG support yang memudahkan pembuatan knowledge base yang selalu up-to-date
Ekosistem connector yang lengkap untuk berbagai sumber data
Developer experience yang baik dengan sintaks mirip pandas

Pathway cocok untuk use case seperti:

Real-time RAG dan chatbot perusahaan
Monitoring dan alerting system
Streaming ETL pipeline
Real-time feature engineering untuk ML
Document processing pipeline

Untuk memulai, cukup instal Pathway dengan pip install pathway dan ikuti contoh-contoh di atas. Dokumentasi lengkap tersedia di website resmi Pathway untuk eksplorasi fitur yang lebih mendalam.