pathwaycom/pathway

Pathway – 實時資料框架

是什麼 – Pathway 是一個開源的 Python 庫，用於建構可在批次與串流資料上運作的資料管道。其底層由基於 Differential Dataflow 的高性能 Rust 引擎執行 Python 定義的邏輯，讓您在使用純 Python 的同時，享受多執行緒、多程序與分散式執行的效能。

對 AI 為何重要 – 此框架內建 LLM x‑pack，包含包裝器、解析器、嵌入器、分割器與記憶體中的向量索引，並支援與 LangChain 及 LlamaIndex 的整合。這讓您能輕鬆建立即時檢索增強生成（RAG）或其他 LLM 驅動的工作流程，對新資料即時反應。

核心功能

• 統一的批次與串流處理 – 一次撰寫管道，即可在靜態檔案或即時串流上執行，無需修改。
• 豐富的連接器生態系 – 內建支援 Kafka、PostgreSQL、GDrive、SharePoint、Airbyte（300+ 資料來源），並支援自訂 Python 連接器。
• 狀態化運算子 – Join、視窗處理、排序與任意 Python UDF 在 Rust 引擎上高速執行。
• 持久化與一致性 – 自動狀態檢查點；免費版提供 至少一次 語意，企業版支援 精確一次。
• 可擴展執行 – 本地多執行緒執行，或於 Docker/Kubernetes 上執行容器；企業版支援分散式雲端部署。
• LLM 輔助工具 – 提供主流 LLM 服務的即用型包裝器、即時向量索引，相容 LangChain/LlamaIndex。

典型應用場景

• 即時 ETL（例如：接收 Kafka 主題，轉換後載入資料庫）。
• 事件驅動的警示管道。
• 持續分析，如即時迴歸或儀表板。
• 即時 LLM/RAG 應用，可即時接收新文件、嵌入並回應查詢。

快速上手

pip install -U pathway   # macOS 或 Linux 上的 Python 3.10+ 支援

一個最小的「正數之和」管道範例：

import pathway as pw

class Input(pw.Schema):
    value: int

src = pw.io.csv.read("./input/", schema=Input)
pos = src.filter(src.value >= 0)
out = pos.reduce(sum_val=pw.reducers.sum(pos.value))

pw.io.jsonlines.write(out, "output.jsonl")

pw.run()

透過 python my_pipeline.py 或 pathway spawn python my_pipeline.py 執行腳本，讓引擎自動管理執行緒。

部署選項

• 本機 – 僅需匯入 pathway 並呼叫 pw.run()。
• Docker – 使用官方 pathwaycom/pathway 鏡像，或在任意 Python 基底鏡像中透過 pip 安裝。
• Kubernetes / 雲端 – 企業版支援容器擴展；Render 上的快速啟動文件已提供。
• 監控 – 內建儀表板可查看連接器吞吐量與延遲。

效能 – 基準測試顯示，在需要時間連接或迭代演算法的工作負載中，其吞吐量高於 Flink、Spark 與 Kafka Streams。

文件與支援 – 完整文件請見 https://pathway.com/developers/，包含 API 參考、範例範本、Discord 社群與問題追蹤器。

授權 – Business Source License 1.1（非商業用途及大多數商業用途免費）。四年後核心程式碼將重新授權為 Apache 2.0。支援倉儲採用 MIT 授權。

---

總結 – Pathway 讓資料工程師與 AI 開發者能以 Python 編寫以「Rust 速度」執行的管道，同時處理歷史資料與即時資料，並為以 LLM 為中心的 RAG 工作流程提供頂級支援。