Pathway：当 Python 终于有了「实时 Flink」，一场重塑数据工程与 RAG 管道的革命

开篇：数据工程的「实时性焦虑」

如果你是一名数据工程师，大概率经历过这样的焦虑时刻：

凌晨 3 点，生产环境告警响起。你打开监控面板，发现昨晚 10 点的数据批处理任务失败了。由于是 T+1 架构，今天上午所有依赖这份数据的业务报表都无法更新。老板问："为什么不能用实时的数据？"你解释："批处理架构就是这样，得等第二天才能看到前一天的完整数据。"

或者更痛苦的：你花了两周时间搭建了一个 RAG 系统，用户反馈说"搜索结果经常是过期的"。你打开日志一看，向量数据库里的文档还是三天前的版本，因为每次全量重建索引需要 4 个小时，你只能每天凌晨跑一次。用户问："能不能实时更新？"你回答："理论上可以，但需要用 Kafka + Flink + 向量数据库的流式架构，改造成本很高。"

这就是数据工程的「实时性焦虑」：实时架构的门槛太高。

要搭建一套真正可用的实时数据管道，你需要：

• Kafka：消息队列，管理数据流
• Flink 或 Spark Streaming：流处理引擎
• Debezium：CDC 变更数据捕获
• Redis/ClickHouse：实时存储
• 向量化 + LLM：如果要接 RAG

这套技术栈的问题是：

1. 学习曲线陡峭：Flink 需要了解 Java/Scala，窗口、水位线、背压等概念
2. 运维成本高：Kafka 集群、ZooKeeper、资源调度、监控告警
3. 开发效率低：从原型到生产，往往需要数周时间

于是大多数团队选择了妥协：用批处理，接受 T+1。

但 2026 年，这个困境被打破了。

一个 Python 框架横空出世，声称：用 Python 写批处理代码，自动变成实时流处理。没有 Kafka，没有 Flink，不需要 JVM，甚至不需要理解流计算的概念。

它就是 Pathway。

一、Pathway 是什么？为什么它能让 Python 程序员尖叫

1.1 核心定位

Pathway 官方定义自己是：

A Python data processing framework for analytics and AI pipelines over data streams

翻译成人话：用 Python 写数据分析代码，自动变成实时流处理。

它的核心卖点：

1.2 技术原理：Differential Dataflow 的魔法

Pathway 的核心是一个 Rust 编写的执行引擎，底层使用 Differential Dataflow 数据结构。

这不是一个新概念，但它之前一直是学术界玩具。Pathway 把它工程化，变成了生产级框架。

什么是 Differential Dataflow？

传统数据流处理的问题是：每次数据更新，你都要重新计算整个结果。

比如你有一个统计用户活跃数的任务：

总活跃用户 = SELECT COUNT(DISTINCT user_id) FROM events

如果有 100 万条事件，每次新来一条事件，传统做法是：

1. 把新事件插入数据库
2. 重新扫描 100 万 + 1 条数据
3. 重新计算 COUNT

这在实时场景下是灾难性的。

Differential Dataflow 的做法是：

1. 不重新计算整个结果
2. 只计算「增量」部分
3. 新结果 = 旧结果 + 增量

示例：假设当前活跃用户数是 5000，新来了一条事件，用户 ID 是 user_123，而这个用户已经在集合里了，那么增量是 0，新结果还是 5000。如果新用户 ID 是 user_9999（第一次出现），增量是 +1，新结果是 5001。

这就是 增量计算：只处理变化的部分，避免全量重算。

1.3 批流一体：一套代码，两种模式

Pathway 最吸引人的特性是 批流一体：

import pathway as pw

# 定义数据源
data = pw.io.csv.read("users.csv")

# 业务逻辑
result = data.groupby(pw.this.city).reduce(
    count=pw.reducers.count()
)

# 输出
pw.io.jsonlines.write(result, "output.jsonl")

# 运行
pw.run()

这段代码可以：

• 批处理模式：读取本地 CSV 文件，计算，输出到 JSON Lines
• 流处理模式：监听文件变化，每当 CSV 更新，自动重新计算，追加输出

切换方式：只需要改一行：

# 批处理
data = pw.io.csv.read("users.csv", mode="static")

# 流处理
data = pw.io.csv.read("users.csv", mode="streaming", autocommit_duration_ms=1000)

这就是「批流一体」的真谛：同样的业务逻辑代码，部署时选择静态模式就是批处理，选择流式模式就是实时处理。

不需要像 Flink 那样，批用 DataSet API，流用 DataStream API，写两套代码。

二、核心概念：从「表」到「流」的思维转换

2.1 Table：Pathway 的核心数据结构

在 Pathway 中，所有数据都是 Table。

这和 Pandas DataFrame 很像，但有一个关键区别：Table 是动态的。

import pathway as pw

# 创建一个静态 Table
table = pw.debug.table_from_markdown('''
  | name  | age
1 | Alice | 25
2 | Bob   | 30
3 | Carol | 28
''')

# 查询
result = table.select(
    name=pw.this.name,
    age_next_year=pw.this.age + 1
)

pw.debug.compute_and_print(result)

输出：

name   | age_next_year
Alice  | 26
Bob    | 31
Carol  | 29

这看起来和 Pandas 没什么区别。但当我们切换到 流式模式，Table 就会变成「活」的：

# 从 Kafka 读取流数据
table = pw.io.kafka.read(
    rdkafka_settings={
        "group.id": "pathway-consumer",
        "bootstrap.servers": "localhost:9092",
        "auto.offset.reset": "earliest"
    },
    topic="user_events",
    format="json"
)

# 业务逻辑（和批处理完全一样）
result = table.groupby(pw.this.user_id).reduce(
    event_count=pw.reducers.count(),
    first_seen=pw.reducers.min(pw.this.timestamp),
    last_seen=pw.reducers.max(pw.this.timestamp)
)

# 写入 Redis（实时更新）
pw.io.redis.write(result, host="localhost", port=6379, key_name="user_stats")

# 启动流处理
pw.run()

这段代码会：

1. 监听 Kafka 的 user_events Topic
2. 每收到一条消息，增量更新 result
3. 实时写入 Redis

你不需要关心窗口、水位线、背压，Pathway 自动处理了这些。

2.2 Connector：350+ 数据源连接器

Pathway 内置了 350+ 连接器，覆盖主流数据源：

输入源：

• 文件系统：CSV、JSON、Parquet、Avro
• 数据库：PostgreSQL、MySQL、MongoDB
• 消息队列：Kafka、RabbitMQ、Redpanda
• 云存储：S3、GCS、Azure Blob
• CDC：Debezium、MongoDB Change Streams
• API：REST API、GraphQL

输出源：

• 数据库：PostgreSQL、MySQL、MongoDB、Redis
• 消息队列：Kafka、RabbitMQ
• 文件：CSV、JSON、Parquet
• 向量数据库：Pinecone、Weaviate、Qdrant
• 实时 Dashboard：Grafana、Streamlit

示例：实时监听 MongoDB 变更

import pathway as pw

# 连接 MongoDB，监听变更
users = pw.io.mongodb.read(
    uri="mongodb://localhost:27017",
    database="mydb",
    collection="users",
    mode="streaming"  # 流式模式，监听 MongoDB Change Stream
)

# 实时计算每个城市的用户数
city_stats = users.groupby(pw.this.city).reduce(
    user_count=pw.reducers.count()
)

# 写入 PostgreSQL
pw.io.postgres.write(
    city_stats,
    uri="postgresql://user:pass@localhost:5432/mydb",
    table_name="city_stats"
)

pw.run()

2.3 Operator：类 SQL 的操作符

Pathway 提供了类似 SQL 和 Pandas 的操作符：

选择与过滤：

# SELECT name, age FROM users WHERE age > 25
result = users.filter(pw.this.age > 25).select(
    name=pw.this.name,
    age=pw.this.age
)

分组聚合：

# SELECT city, COUNT(*) FROM users GROUP BY city
result = users.groupby(pw.this.city).reduce(
    count=pw.reducers.count()
)

Join：

# SELECT * FROM orders JOIN users ON orders.user_id = users.id
result = orders.join(
    users,
    pw.left.user_id == pw.right.id
).select(
    order_id=pw.left.id,
    user_name=pw.right.name,
    amount=pw.left.amount
)

窗口函数：

# 滑动窗口：过去 1 小时的事件数
import datetime

result = events.filter(
    pw.this.timestamp >= pw.this.timestamp - datetime.timedelta(hours=1)
).groupby(
    pw.this.user_id
).reduce(
    event_count=pw.reducers.count()
)

Temporal Join（时间 Join）：

# 基于时间的 Join：订单关联当时的汇率
result = orders.join_left(
    exchange_rates,
    pw.left.currency == pw.right.currency,
    pw.left.timestamp >= pw.right.valid_from,
    pw.left.timestamp < pw.right.valid_to
).select(
    order_id=pw.left.id,
    amount=pw.left.amount,
    rate=pw.right.rate
)

三、实战：从零搭建实时 ETL 管道

3.1 场景：实时用户行为分析

假设我们要搭建一个实时用户行为分析系统，需求：

1. 从 Kafka 读取用户行为日志（点击、浏览、购买）
2. 实时统计每个用户的行为数、活跃时长
3. 实时计算每个页面的 PV/UV
4. 结果写入 PostgreSQL，供 Dashboard 展示

传统方案：Kafka + Flink + Redis + PostgreSQL，开发周期约 2 周。

Pathway 方案：单文件 Python 脚本，开发时间 1 天。

3.2 完整代码

import pathway as pw
from datetime import datetime, timedelta

# 1. 定义 Kafka 数据源
events = pw.io.kafka.read(
    rdkafka_settings={
        "group.id": "user-analytics",
        "bootstrap.servers": "localhost:9092",
        "auto.offset.reset": "latest"
    },
    topic="user_events",
    format="json",
    schema=pw.schema_from_types(
        user_id=str,
        event_type=str,  # click, view, purchase
        page_url=str,
        timestamp=int
    )
)

# 2. 数据清洗与转换
events_clean = events.select(
    user_id=pw.this.user_id,
    event_type=pw.this.event_type,
    page_url=pw.this.page_url,
    timestamp=pw.this.timestamp,
    datetime=pw.make_datetime(pw.this.timestamp)  # Unix 时间戳转 datetime
)

# 3. 实时统计：每个用户的行为数
user_stats = events_clean.groupby(pw.this.user_id).reduce(
    total_events=pw.reducers.count(),
    first_seen=pw.reducers.min(pw.this.datetime),
    last_seen=pw.reducers.max(pw.this.datetime),
    click_count=pw.reducers.sum(pw.cast(int, pw.this.event_type == "click")),
    view_count=pw.reducers.sum(pw.cast(int, pw.this.event_type == "view")),
    purchase_count=pw.reducers.sum(pw.cast(int, pw.this.event_type == "purchase"))
)

# 4. 计算活跃时长（秒）
user_stats = user_stats.select(
    user_id=pw.this.user_id,
    total_events=pw.this.total_events,
    click_count=pw.this.click_count,
    view_count=pw.this.view_count,
    purchase_count=pw.this.purchase_count,
    active_seconds=pw.cast(int, pw.this.last_seen - pw.this.first_seen)
)

# 5. 实时统计：每个页面的 PV/UV
page_stats = events_clean.groupby(pw.this.page_url).reduce(
    pv=pw.reducers.count(),  # Page View
    uv=pw.reducers.count_distinct(pw.this.user_id)  # Unique Visitor
)

# 6. 滑动窗口：过去 1 小时的实时统计
recent_events = events_clean.filter(
    pw.this.datetime >= datetime.utcnow() - timedelta(hours=1)
)

recent_user_stats = recent_events.groupby(pw.this.user_id).reduce(
    recent_events=pw.reducers.count()
)

# 7. 输出到 PostgreSQL
pw.io.postgres.write(
    user_stats,
    uri="postgresql://analytics:password@localhost:5432/analytics_db",
    table_name="user_stats"
)

pw.io.postgres.write(
    page_stats,
    uri="postgresql://analytics:password@localhost:5432/analytics_db",
    table_name="page_stats"
)

# 8. 启动流处理
pw.run()

3.3 代码解析

关键点 1：Schema 定义

schema=pw.schema_from_types(
    user_id=str,
    event_type=str,
    page_url=str,
    timestamp=int
)

Pathway 需要显式定义 Schema，这和 Pandas 不同。好处是：

• 编译时类型检查
• 优化查询计划
• 自动推断列类型

关键点 2：条件聚合

click_count=pw.reducers.sum(pw.cast(int, pw.this.event_type == "click"))

这里用了一个技巧：pw.this.event_type == "click" 返回布尔值，pw.cast(int, ...) 转成 0/1，然后用 sum 累加。

等价于 SQL：

SUM(CASE WHEN event_type = 'click' THEN 1 ELSE 0 END)

关键点 3：时间差计算

active_seconds=pw.cast(int, pw.this.last_seen - pw.this.first_seen)

两个 datetime 相减，得到 timedelta，Pathway 会自动处理。

关键点 4：滑动窗口

recent_events = events_clean.filter(
    pw.this.datetime >= datetime.utcnow() - timedelta(hours=1)
)

这里用 filter + 时间条件实现滑动窗口。Pathway 会自动管理窗口状态，旧数据会被清理。

3.4 部署

Pathway 支持多种部署方式：

1. 直接运行：

python user_analytics.py

2. Docker：

FROM python:3.11-slim

RUN pip install pathway

COPY user_analytics.py /app/
WORKDIR /app

CMD ["python", "user_analytics.py"]

3. Kubernetes：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: pathway-analytics
spec:
  replicas: 3
  selector:
    matchLabels:
      app: pathway-analytics
  template:
    metadata:
      labels:
        app: pathway-analytics
    spec:
      containers:
      - name: pathway
        image: pathway/analytics:latest
        env:
        - name: KAFKA_BROKERS
          value: "kafka:9092"

四、高级特性：RAG 管道与 LLM 集成

4.1 为什么 Pathway 适合 RAG

RAG（Retrieval-Augmented Generation）是当前最流行的 LLM 应用架构。但传统 RAG 有一个致命问题：知识库更新不及时。

典型 RAG 流程：

1. 离线：文档 → Embedding → 向量数据库
2. 在线：查询 → 向量检索 → LLM 生成

问题在于第 1 步：文档更新后，需要重新 Embedding、重新写入向量数据库。这个过程通常很慢（几小时到几天），导致 RAG 系统的知识是「过时」的。

Pathway 的解决方案：实时 RAG 管道。

4.2 实时 RAG 架构

import pathway as pw
from pathway.xpacks.llm.parsers import UnstructuredParser
from pathway.xpacks.llm.splitters import TokenCountSplitter
from pathway.xpacks.llm.embedders import OpenAIEmbedder
from pathway.xpacks.llm.vector_store import VectorStoreServer
import os

# 1. 监听文档目录（流式模式）
files = pw.io.fs.read(
    os.environ.get("DOCS_DIR", "./docs"),
    mode="streaming",  # 监听文件变化
    format="binary",
    autocommit_duration_ms=1000  # 每秒检查一次
)

# 2. 解析文档（支持 PDF、Word、HTML 等）
parser = UnstructuredParser(chunking_mode="elements")
documents = files.select(
    elements=parser(pw.this.data)
)

# 3. 展平解析结果
documents = documents.flatten(pw.this.elements)
documents = documents.select(
    text=pw.this.elements[0],
    metadata=pw.this.elements[1]
)

# 4. 切分文本
splitter = TokenCountSplitter(min_tokens=100, max_tokens=500)
chunks = documents.select(
    chunk=splitter(pw.this.text),
    metadata=pw.this.metadata
)

chunks = chunks.flatten(pw.this.chunk)
chunks = chunks.select(
    text=pw.this.chunk[0],
    metadata=pw.this.chunk[1]
)

# 5. 生成 Embedding
embedder = OpenAIEmbedder(model="text-embedding-3-small")
embeddings = chunks.select(
    text=pw.this.text,
    metadata=pw.this.metadata,
    embedding=embedder(pw.this.text)
)

# 6. 构建向量索引（实时更新）
vector_store = VectorStoreServer(
    embeddings,
    embedder=embedder,
    host="0.0.0.0",
    port=8765
)

# 7. 启动服务
pw.run()

关键特性：

• 实时更新：新增/修改/删除文档，向量索引自动更新
• 增量计算：只重新计算变化的部分，不需要全量重建
• API 服务：VectorStoreServer 提供 REST API，供查询

4.3 查询 RAG 服务

import requests

# 查询接口
response = requests.post(
    "http://localhost:8765/v1/query",
    json={
        "query": "What is Pathway?",
        "k": 5
    }
)

results = response.json()
for hit in results["hits"]:
    print(f"Score: {hit['score']}")
    print(f"Text: {hit['text'][:200]}...")
    print(f"Metadata: {hit['metadata']}")
    print("---")

4.4 集成 LangChain / LlamaIndex

Pathway 提供了开箱即用的集成：

LangChain 集成：

from langchain.vectorstores import PathwayVectorClient
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 连接 Pathway Vector Store
vectorstore = PathwayVectorClient(
    url="http://localhost:8765"
)

# 构建 RAG Chain
llm = ChatOpenAI(model="gpt-4o")
qa = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# 查询
answer = qa.run("What is Differential Dataflow?")
print(answer)

LlamaIndex 集成：

from llama_index.vector_stores.pathway import PathwayVectorStore
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.llms.openai import OpenAI

# 连接 Pathway
vector_store = PathwayVectorStore(
    client_url="http://localhost:8765"
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 构建索引
index = VectorStoreIndex.from_vector_store(
    vector_store,
    storage_context=storage_context
)

# 查询
llm = OpenAI(model="gpt-4o")
query_engine = index.as_query_engine(llm=llm, similarity_top_k=5)
response = query_engine.query("Explain Pathway's architecture")
print(response)

4.5 Reranker：提升检索精度

Pathway 内置了多种 Reranker：

from pathway.xpacks.llm.rerankers import CrossEncoderReranker, LLMReranker
from pathway.xpacks.llm.question_answering import BaseRAGQuestionAnswerer

# 方法 1：Cross-Encoder Reranker（基于 BERT）
reranker = CrossEncoderReranker(model="cross-encoder/ms-marco-MiniLM-L-6-v2")

# 方法 2：LLM Reranker（让 LLM 打分）
reranker = LLMReranker(model="gpt-4o-mini")

# 构建 RAG
rag = BaseRAGQuestionAnswerer(
    llm_model="gpt-4o",
    indexer=vector_store,
    reranker=reranker,
    search_topk=20,  # 先召回 20 个
    rerank_topk=5    # Rerank 后取前 5 个
)

# 启动 RAG 服务
rag_server = rag.build_server(host="0.0.0.0", port=8766)
pw.run()

Reranker 的作用：

1. 先用向量检索召回 Top 20（速度快）
2. 再用 Reranker 精排，取 Top 5（精度高）
3. 两阶段检索，兼顾效率与效果

五、性能对比：Pathway vs Flink vs Spark Streaming

5.1 基准测试环境

• 数据源：Kafka（100 万条事件/秒）
• 集群：5 节点，每节点 8 核 16GB
• 任务：实时聚合（GroupBy + Count + Sum）

5.2 结果

结论：

• Pathway 在吞吐量和延迟上略逊于 Flink，但差距不大
• 内存占用最低，适合边缘计算
• 开发效率最高，从原型到生产只需 1 天

5.3 适用场景

适合 Pathway：

• 中小规模实时 ETL（< 1M events/sec）
• RAG 管道、实时搜索
• Python 技术栈团队
• 快速原型验证

适合 Flink：

• 大规模实时计算（> 1M events/sec）
• 复杂事件处理（CEP）
• Java/Scala 技术栈团队
• 对延迟极度敏感（< 50ms）

适合 Spark Streaming：

• 批流统一（已有 Spark 批处理）
• 大规模批处理 + 微批流处理
• 机器学习管道（MLlib）

六、生产实践踩坑

6.1 背压处理

流处理系统常见问题：下游消费慢于上游生产，导致数据积压。

Flink 的解决方案是背压机制：上游检测到下游消费慢，自动降低生产速度。

Pathway 的处理方式不同：Differential Dataflow 自动处理背压。

原理：由于是增量计算，每次只处理变化的部分，内存占用是 O(变化量) 而不是 O(总量)。即使上游数据积压，只要增量控制在合理范围，系统不会崩溃。

最佳实践：

# 设置 autocommit_duration_ms，控制提交频率
data = pw.io.kafka.read(
    ...,
    autocommit_duration_ms=5000  # 每 5 秒提交一次
)

6.2 Exactly-Once 语义

流处理的难题：数据可能重复消费。

• Kafka 至少一次投递
• 消费者可能重启，重新消费消息
• 导致结果重复

Pathway 通过 幂等写入 保证 Exactly-Once：

# PostgreSQL 幂等写入
pw.io.postgres.write(
    result,
    ...,
    primary_key=["user_id"],  # 指定主键
    on_conflict="update"       # 冲突时更新
)

6.3 状态持久化

流处理任务重启后，需要恢复状态。Pathway 支持 状态持久化：

# 配置持久化
pw.run(
    persistence_engine=pw.persistence.Backend(
        backend_type="filesystem",
        path="./pathway_state"
    )
)

重启后，Pathway 会自动恢复状态，从上次的位置继续消费。

6.4 监控与可观测性

Pathway 原生支持 OpenTelemetry：

import pathway as pw

# 配置监控
pw.set_monitoring_config(
    enable_metrics=True,
    enable_tracing=True,
    otlp_endpoint="http://otel-collector:4317"
)

# Prometheus 指标
# - pathway_rows_processed_total
# - pathway_latency_seconds
# - pathway_memory_usage_bytes

七、生态与社区

7.1 生态

7.2 社区

• GitHub Stars：62k+
• Discord 社区：活跃
• 文档质量：优秀（官方教程、API 文档、示例代码）
• 企业客户：多家知名公司采用

八、总结：Pathway 的定位与未来

8.1 定位

Pathway 不是要取代 Flink/Spark，而是要 降低实时计算的门槛。

它的核心价值：

1. Python 优先：让 Python 程序员也能做实时计算
2. 批流一体：一套代码，两种模式
3. AI 原生：内置 RAG、Embedding、向量索引

8.2 适用场景

8.3 学习资源

• 官方文档：https://pathway.com/developers/
• GitHub：https://github.com/pathwaycom/pathway
• 模板库：https://pathway.com/developers/templates
• Discord 社区：https://discord.gg/pathway

8.4 未来展望

Pathway 正在快速发展：

• 性能优化：进一步逼近 Flink 的吞吐量
• 生态扩展：更多连接器、更多 LLM 集成
• 云服务：Pathway Cloud 托管服务

一句话总结：Pathway 是 Python 程序员进入实时计算领域的最佳入口，也是 RAG 管道的理想选择。如果你曾被 Flink 的复杂劝退，Pathway 值得一试。

附录：完整代码仓库

本文所有示例代码已上传 GitHub：

git clone https://github.com/pathwaycom/pathway-examples
cd pathway-examples/realtime-etl
python -m venv venv
source venv/bin/activate
pip install pathway
python user_analytics.py

运行前提：

• Python 3.9+
• Kafka 集群（或使用 Docker Compose）
• PostgreSQL（可选，用于输出）

字数统计：约 6,500 字

标签：Pathway | Python | 实时计算 | ETL | RAG | Differential Dataflow | 流处理 | 数据工程

关键词：Pathway, Python, real-time ETL, streaming, RAG, Differential Dataflow, Kafka, vector store, LLM, data engineering