Getting Started | Deployment | Documentation and Support | Blog | License

Pathway 实时数据框架

Pathway 是一个用于流处理、实时分析、LLM 流水线和 RAG 的 Python ETL 框架。

Pathway 带有一个易于使用的 Python API，使您能够无缝集成最喜爱的 Python 机器学习库。Pathway 代码用途广泛且稳健：您可以在开发和生产环境中使用它，高效处理批处理和流式数据。同一段代码可用于本地开发、CI/CD 测试、运行批处理作业、处理流式回放以及实时数据流。

Pathway 由基于 Differential Dataflow 的可扩展 Rust 引擎提供支持，并执行增量计算。尽管您的 Pathway 代码是用 Python 编写的，但它实际上是由 Rust 引擎执行的，从而实现多线程、多进程和分布式计算。整个流水线都保存在内存中，可以轻松地通过 Docker 和 Kubernetes 部署。

您可以通过 pip 安装 Pathway：

pip install -U pathway

如有任何问题，您可以在 Discord 上找到社区和项目团队。

使用场景与模板

准备好看看 Pathway 能做什么了吗？

尝试我们易于运行的示例！

这些示例以笔记本和 Docker 格式提供，只需点击几下即可启动。选择一个，立即开始您的 Pathway 实践体验吧！

事件处理与实时分析流水线

凭借其适用于批处理和流式的统一引擎以及完全的 Python 兼容性，Pathway 使数据处理尽可能简单。它是多种数据处理流水线的理想解决方案，包括：

• 展示：实时 ETL。
• 展示：带有警报的事件驱动流水线。
• 展示：实时分析。
• 文档：从批处理切换到流式处理。

AI 流水线

Pathway 提供专门的 LLM 工具，用于构建实时 LLM 和 RAG 流水线。其中包含了对大多数常见 LLM 服务的封装和实用工具，使得处理 LLM 和 RAG 流水线变得异常简单。请查看我们的 LLM xpack 文档。

不妨试试我们提供的包含 LLM 工具的可运行示例。 您可以在这里找到这些示例。

• 模板：非结构化数据到 SQL 的即时转换。
• 模板：使用 Ollama 和 Mistral AI 的私有 RAG
• 模板：自适应 RAG
• 模板：结合 gpt-4o 的多模态 RAG

功能特性

• 丰富的连接器：Pathway 自带多种连接器，可对接 Kafka、GDrive、PostgreSQL 或 SharePoint 等外部数据源。借助 Airbyte 连接器，您还可以接入超过 300 种不同的数据源。如果所需连接器尚未提供，您也可以使用 Pathway 的 Python 连接器框架构建自定义连接器。
• 无状态与有状态转换：Pathway 支持有状态转换，例如连接、窗口计算和排序等操作。它内置了大量由 Rust 直接实现的转换功能。除了这些预置功能外，您还可以使用任意 Python 函数来处理数据——无论是自定义函数，还是调用任何 Python 库。
• 持久化支持：Pathway 提供持久化机制，用于保存计算状态。这使得您在更新或系统崩溃后能够轻松重启管道。您的数据管道将由 Pathway 安全可靠地管理！
• 一致性保障：Pathway 会自动处理时间相关问题，确保所有计算的一致性。特别是对于迟到或乱序的数据点，Pathway 会在新数据到达时及时更新结果。免费版提供“至少一次”一致性保证，而企业版则提供“恰好一次”一致性。
• 可扩展的 Rust 引擎：借助 Pathway 的 Rust 引擎，您可以摆脱 Python 常见的性能限制，轻松实现多线程、多进程以及分布式计算。
• 大模型助手：Pathway 提供 LLM 扩展模块，包含集成大模型到数据管道所需的所有工具（LLM 封装器、解析器、嵌入器、分词器），还内置了内存中的实时向量索引，并与 LLamaIndex 和 LangChain 等框架无缝集成。您可以快速构建并部署基于实时文档的 RAG 应用程序。

开始使用

安装

Pathway 需要 Python 3.10 或更高版本。

您可以通过 pip 安装最新版本的 Pathway：

$ pip install -U pathway

⚠️ Pathway 目前仅支持 macOS 和 Linux 系统。其他操作系统用户建议在虚拟机中运行 Pathway。

示例：实时计算正数之和。

import pathway as pw

# 定义数据模式（可选）
class InputSchema(pw.Schema):
  value: int

# 使用连接器接入数据
input_table = pw.io.csv.read(
  "./input/",
  schema=InputSchema
)

# 定义数据处理逻辑
filtered_table = input_table.filter(input_table.value>=0)
result_table = filtered_table.reduce(
  sum_value = pw.reducers.sum(filtered_table.value)
)

# 将结果输出到外部系统
pw.io.jsonlines.write(result_table, "output.jsonl")

# 启动计算
pw.run()

您可以在 Google Colab 中运行此示例。

更多示例请参阅 GitHub 仓库。

部署

本地部署

使用 Pathway 时，只需导入即可：

import pathway as pw

随后，您可以轻松构建数据处理管道，让 Pathway 自动处理数据更新。管道搭建完成后，只需一条命令即可开始对流式数据进行计算：

pw.run()

之后，您就可以像普通 Python 脚本一样运行 Pathway 项目（例如 main.py）：$ python main.py。

Pathway 内置监控仪表板，可跟踪各连接器发送的消息数量及系统延迟情况，同时还显示日志信息。

此外，您也可以使用 pathway'ish 版本：

$ pathway spawn python main.py

Pathway 原生支持多线程。若希望以 3 个线程运行应用，可执行以下命令：

$ pathway spawn --threads 3 python main.py

为快速启动 Pathway 项目，您可以使用我们的 cookiecutter 模板。

Docker 部署

您可以通过 Docker 轻松运行 Pathway。

Pathway 容器镜像

您可以使用 Pathway Docker 镜像，并配合以下 Dockerfile：

FROM pathwaycom/pathway:latest

WORKDIR /app

COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD [ "python", "./your-script.py" ]

然后构建并运行该镜像：

docker build -t my-pathway-app .
docker run -it --rm --name my-pathway-app my-pathway-app

运行单个 Python 脚本

对于仅包含一个文件的项目，创建完整的 Dockerfile 可能显得多余。在这种情况下，您可以直接使用 Pathway Docker 镜像运行 Python 脚本。例如：

docker run -it --rm --name my-pathway-app -v "$PWD":/app pathwaycom/pathway:latest python my-pathway-app.py

标准 Python 镜像

您也可以使用标准 Python 镜像，并通过 Dockerfile 安装 Pathway：

FROM --platform=linux/x86_64 python:3.10

RUN pip install -U pathway
COPY ./pathway-script.py pathway-script.py

CMD ["python", "-u", "pathway-script.py"]

Kubernetes 与云部署

Docker 容器非常适合在 Kubernetes 上部署于云端。如果您需要扩展 Pathway 应用程序规模，可以考虑我们的 Pathway for Enterprise 版本。Pathway for Enterprise 专为端到端数据处理和实时智能分析设计，支持云端分布式计算与 Kubernetes 分布式部署，并可配置外部持久化存储。

您可以通过 Render 等服务轻松部署 Pathway：请参阅 如何通过几下点击部署 Pathway。

如需了解更多信息，请随时联系我们：[contact@pathway.com]。

性能

Pathway 的设计目标是超越当前最先进的流式与批处理技术，包括 Flink、Spark 和 Kafka Streaming 等。它还允许在流式场景下实现许多其他流处理框架难以直接支持的算法或自定义函数（例如：时间窗口连接、迭代图算法、机器学习流程）。

如果您感兴趣，可以参考以下基准测试结果：Pathway 基准测试库。

文档与支持

Pathway 的完整文档可在 pathway.com/developers/ 查阅，其中包括 API 文档。

如果您有任何疑问，请随时在 GitHub 上提交 issue，加入我们的 Discord 社区，或发送邮件至 contact@pathway.com。

🤝 重点合作与集成

我们构建前沿的数据处理管道，并联合推广那些突破 Python 和流式数据应用边界的技术解决方案。以下是与我们携手塑造数据工程未来的重要伙伴。

许可证

Pathway 采用 BSL 1.1 许可证 进行分发，该许可证允许无限制的非商业用途，以及在大多数商业场景下免费使用 Pathway 软件包。本仓库中的代码将在 4 年后自动转为开源（Apache 2.0 许可证）。此外，一些与本仓库互补的 公共仓库（如示例、库、连接器等）则以 MIT 许可证作为开源许可。

贡献指南

如果您正在开发希望与本仓库集成的库或连接器，建议您先将其作为独立仓库，采用 MIT 或 Apache 2.0 许可证发布。

关于 Pathway 核心功能的所有问题，欢迎提交 Issue。如需更多信息，请随时加入 Pathway 的 Discord 社区。

Pathway 快速上手指南

Pathway 是一个用于流处理、实时分析、LLM 管道和 RAG 的 Python ETL 框架。它拥有易用的 Python API，底层由基于 Differential Dataflow 的可扩展 Rust 引擎驱动，支持多线程、多进程及分布式计算，能够无缝处理批处理和流式数据。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：支持 macOS 和 Linux。
  • 注意：如果您使用的是 Windows 或其他系统，建议在虚拟机（VM）或 WSL (Windows Subsystem for Linux) 中运行。
• Python 版本：Python 3.10 或更高版本。
• 前置依赖：建议安装 pip 包管理工具。

安装步骤

您可以直接使用 pip 安装最新版本的 Pathway。

pip install -U pathway

提示：如果遇到网络下载缓慢的问题，可以使用国内镜像源加速安装：

pip install -U pathway -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

Pathway 的核心优势在于同一套代码即可用于本地开发、测试，也可用于生产环境的流式处理。以下是一个最简单的示例：实时计算输入数据中正数的总和。

示例代码

创建一个名为 main.py 的文件，并写入以下内容：

import pathway as pw

# 1. 定义数据模式 (可选)
class InputSchema(pw.Schema):
  value: int

# 2. 使用连接器读取数据 (此处以读取本地 CSV 文件夹为例)
input_table = pw.io.csv.read(
  "./input/",
  schema=InputSchema
)

# 3. 定义数据处理逻辑
# 过滤出大于等于 0 的值
filtered_table = input_table.filter(input_table.value>=0)
# 对过滤后的值进行求和聚合
result_table = filtered_table.reduce(
  sum_value = pw.reducers.sum(filtered_table.value)
)

# 4. 将结果输出到外部系统 (此处输出为 JSONL 文件)
pw.io.jsonlines.write(result_table, "output.jsonl")

# 5. 运行计算管道
pw.run()

运行项目

准备好输入数据（在 ./input/ 目录下放置 CSV 文件）后，像运行普通 Python 脚本一样执行：

python main.py

或者使用 Pathway 自带的启动命令（支持原生多线程）：

pathway spawn python main.py

若需指定线程数（例如 3 个线程），可执行：

pathway spawn --threads 3 python main.py

运行后，Pathway 会自动监控数据源的变化，实时增量更新计算结果，并可通过内置的监控仪表盘查看系统状态。