7.1K+ Star！Pathway：一个实时数据处理ETL框架

小半 • 2025年2月15日下午10:01 • 技术前沿 • 阅读 87

欢迎关注我，持续获取更多内容，感谢赞&在看~

Pathway 简介

Pathway^[1] 是一个Python ETL（Extract-Transform-Load）框架，专注于流处理、实时分析、大型语言模型（LLM）管道和检索-生成（RAG）应用。它提供了一个易于使用的Python API，可以无缝集成你最喜欢的Python机器学习库。

Pathway代码具有通用性和鲁棒性，可以在开发和生产环境中使用，有效处理批处理和流数据。同样的代码可以用于本地开发、CI/CD测试、运行批处理作业、处理流重放和数据流处理。

项目特点

主要特点

易用的Python API：无缝集成Python机器学习库。
批处理与流处理统一引擎：简化数据处理流程。
强大的Rust引擎：基于Differential Dataflow，执行增量计算，支持多线程、多进程和分布式计算。
内存中的数据管道：易于使用Docker和Kubernetes部署。
连接器丰富：支持连接到Kafka、GDrive、PostgreSQL或SharePoint等外部数据源。
无状态和有状态转换：支持诸如连接、窗口和排序等有状态转换。
持久性：提供计算状态的保存，以便在更新或崩溃后重启管道。
一致性：自动处理时间，确保所有计算一致性，管理迟到和乱序的数据点。
LLM助手：提供LLM扩展，包含所有实用工具以将LLM集成到数据管道中。

使用场景

Pathway适用于多种数据处理管道，包括：

实时ETL
事件驱动的管道与告警
实时分析
从批处理转换为流处理

项目使用

安装

Pathway需要Python 3.10或以上版本。可以使用pip安装：

pip install -U 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()