Faust 入门指南：用 Python 与 Kafka 构建分布式流处理应用的完整介绍

• 流处理
• 消息队列
• 后端

【免费下载链接】faust

Python Stream Processing

项目地址：
https://gitcode.com/gh_mirrors/fa/faust

点击查看 免费下载

本指南基于 Faust 官方文档的 Introduction 章节，系统讲解这个 Python 流处理库的核心能力（Agent、Table、异步并发）、典型应用场景、环境要求、扩展包体系与设计理念，并结合仓库源码（如 faust/agents/agent.py、faust/streams.py、examples/word_count.py）给出可复现的示例与底层实现佐证。读完本文，你将理解 Faust 如何用纯 Python 语法处理无限数据流、如何在多机多核上水平扩展有状态计算，并掌握从环境准备、事件接入到编写第一个处理拓扑的完整路径。

Faust 是什么，它能做什么

Faust 是一个用于构建分布式流处理系统的 Python 库，核心定位是“Python Stream Processing”。它与 Kafka 深度集成，但抽象层面足够通用，也可以用于基于 Agent 与 Channel 的通用异步编程（见 faust/init.py 顶部模块说明）。Faust 的三大核心能力分别是 Agent、Table 与基于 asyncio 的异步并发。

核心能力一：Agent —— 用异步生成器处理无限流

Agent 是 Faust 最基本的处理单元：它本质是一个异步生成器函数，通过 async for 遍历流中的每条事件。Agent 的概念源自 Actor 模型，这意味着同一个流处理器可以：

• 在单台机器的多个 CPU 核上并发执行；
• 在数百台机器上同时运行同一套处理逻辑。

处理流的代码就是常规 Python 语法，因此你可以直接复用自己熟悉的第三方库：

@app.agent()
async def process(stream):
async for value in stream:
process(value)

从源码看，@app.agent 装饰器返回的是 Agent 对象（定义于 faust/agents/agent.py），它继承自 mode 库的 Service，并通过 concurrency 参数控制单进程内并发 Actor 数量；当 isolated_partitions=True 时，每个分区由独立的 Actor 处理（此时 concurrency 必须为 1，否则会抛出 ImproperlyConfigured 异常）。Agent 还支持在装饰器中声明 sink 或在运行期通过 add_sink() 追加下游处理，也内置了 ask/cast 等 RPC 风格调用（ask 会等待处理结果返回，cast 则发送后即返回），这些能力使其天然适合构建复杂的流拓扑。

核心能力二：Table —— 分片字典让处理有状态

大多数流处理场景需要状态：比如统计订单、维护计数、做窗口聚合。Faust 的 Table 是分片化的字典，它让流处理器可以持有持久化、可恢复的状态：

• Streams 被分区存放，使相关数据保持在“就近”的节点上；
• 可以轻松地重分区（repartition），拼装出你想要的拓扑结构。

下面的例子演示了如何把订单流按 account_id 重分区，并在分布式表中统计每个账户的订单数：

import faust

# 这个模型描述了消息值在 Kafka "orders" topic 中的序列化方式。
class Order(faust.Record, serializer='json'):
account_id: str
product_id: str
amount: int
price: float

app = faust.App('hello-app', broker='kafka://localhost')
orders_kafka_topic = app.topic('orders', value_type=Order)

# 我们的表在所有 worker 实例间分片，并带 standby 副本，
# 当某个节点宕机时 standby 节点会接管。
order_count_by_account = app.Table('order_count', default=int)

@app.agent(orders_kafka_topic)
async def process(orders: faust.Stream[Order]) -> None:
async for order in orders.group_by(Order.account_id):
order_count_by_account[order.account_id] += 1

这里的关键是 group_by(Order.account_id)：它基于字段描述符生成新 key 并触发重分区。从 faust/streams.py 的实现看，group_by 可以接收字段描述符、普通可调用对象或异步可调用对象作为 key 提取器（若 key 是 callable，则必须通过 name 参数指定重分区 topic 的后缀），底层会创建内部 topic 并重新按键分发消息。

因此，当你在多台机器上启动多个 Faust 应用实例时，相同 account_id 的订单一定会被投递给同一个流处理 Agent，从而保证表内计数正确更新。分片/分区是有状态流处理应用的基石，设计系统时必须把这一点纳入考量；不过流也支持 round-robin 顺序处理，因此 Faust 同样可以用于事件处理和任务队列。

核心能力三：基于 asyncio 的异步并发

Faust 充分利用了 Python 3.6+ 的 asyncio 与 async/await 语法，让多个流处理器可以在同一个进程内并行运行，并可以同时承载 Web 服务器等其它网络服务。借助 Faust 与 asyncio，你可以把流处理拓扑直接嵌入到已有的 asyncio、eventlet、Twisted、Tornado 应用之中，而不必单独部署一套独立的流处理集群。

Faust 的设计定位（Faust is…）

官方文档用五个词概括了 Faust 的设计取向：

Simple（简单） Faust 极易上手。使用其它流处理方案往往需要复杂的 hello-world 工程和基础设施要求，而 Faust 只依赖 Kafka，其余部分就是纯 Python。你只要会 Python，就能用 Faust 做流处理，并能与几乎任何系统集成。下面是官方文档给出的最简单示例之一：

import faust

class Greeting(faust.Record):
from_name: str
to_name: str

app = faust.App('hello-app', broker='kafka://localhost')
topic = app.topic('hello-topic', value_type=Greeting)

@app.agent(topic)
async def hello(greetings):
async for greeting in greetings:
print(f'Hello from {greeting.from_name} to {greeting.to_name}')

@app.timer(interval=1.0)
async def example_sender(app):
await hello.send(
value=Greeting(from_name='Faust', to_name='you'),
)

if __name__ == '__main__':
app.main()

这个示例应用启动了两个任务：一个是处理流的 Agent，另一个是每隔 1 秒向该流发送事件的后台定时任务（@app.timer）。在真实业务中，事件通常由你的系统发布到 Kafka topic 供处理器消费，这里的定时发送只是为了喂数据给示例。你不需要深刻理解 asyncio 的原理——模仿示例写法即可正常使用。

Highly Available（高可用） Faust 可以承受网络问题与服务器崩溃：节点故障时能自动恢复，Table 配备的 standby 副本会在主节点失效时接管。

Distributed（分布式） 按需启动更多应用实例即可水平扩展。

Fast（高性能） 官方文档指出，单个 CPU 核的 Faust worker 实例每秒即可处理数万条事件（tens of thousands of events every second），并且随着更优化的 Kafka 客户端被支持，吞吐量还有进一步提升空间。

Flexible（灵活） Faust 就是 Python：一条流就是一个无限异步迭代器。会 Python 就等于会 Faust，并且可以与 Django、Flask、SQLAlchemy、NLTK、NumPy、SciPy、TensorFlow 等常用 Python 库协同工作。

Faust 的典型应用场景

官方文档列举了 Faust 的主要用途（分为两栏）：

• 事件处理（Event Processing）
• 分布式 Join 与聚合（Distributed Joins & Aggregations）
• 机器学习（Machine Learning）
• 异步任务（Asynchronous Tasks）
• 分布式计算（Distributed Computing）
• 数据反规范化（Data Denormalization）
• 入侵检测（Intrusion Detection）
• 实时 Web 与 WebSocket（Realtime Web & Web Sockets）

以及“更多其他场景”。仓库中对应着丰富的真实示例可供参考，例如 examples/word_count.py（分布式单词计数，含 Table 聚合与 Web 页面查询）、examples/windowed_aggregation.py（窗口聚合）、examples/withdrawals.py（流水事件处理）与 examples/leader.py（选举/分布式协调）等。

怎么用：两步走

Step 1：把事件接入你的系统

一切从事件源开始——无论发生了什么业务事件，都发布到 Kafka：

• 创建了账户？发布到 Kafka。
• 用户修改了密码？发布到 Kafka。
• 有人下单、发表评论、打了标签……？全部发布到 Kafka！

Step 2：用 Faust 处理这些事件

基于上面的事件，你可以写出各种处理逻辑，例如：

• 订单发货时发送邮件通知；
• 找出连续三天以上没有对应发货事件的订单；
• 发现从可疑 IP 地址修改密码的账户。

本质模式是：业务系统负责把事件源源不断写入 Kafka topic，Faust 应用则订阅这些 topic 并执行你定义的异步处理函数。

环境要求

Faust 需要 Python 3.6 或更高版本，以及一个正在运行的 Kafka broker（Kafka 0.10.1 或更高版本）。官方明确表示没有计划支持更早的 Python 版本。

这一约束同样体现在打包配置中：见 setup.py，其中 python_requires='>=3.6.0'，并且在导入阶段若 sys.version_info < (3, 6) 会直接抛出异常。文档还提到，若你需要 RocksDB 持久化存储，则要求 RocksDB 5.0 或更高版本并安装 python-rocksdb。

扩展包（Extras）体系一览

Faust 通过 setuptools extras 机制提供可选依赖，可以按需组合安装。

核心扩展

性能优化扩展

以下扩展可以用一条命令全部安装：pip install faust[fast]

其中 cython、orjson、ciso8601 等主要用于加速关键路径：仓库在 setup.py 中会尝试将 faust/_cython/windows.pyx、faust/_cython/streams.pyx、faust/transport/_cython/conductor.pyx 编译为 C 扩展（编译失败时自动回退到纯 Python 实现），而 orjson 可加速 JSON 序列化/反序列化。

调试扩展

以下扩展可以用一条命令全部安装：pip install faust[debug]

组合安装与特殊平台注意点

• 同时安装多个扩展：用逗号分隔即可，例如 pip install faust[uvloop,fast,rocksdb,datadog,redis]。
• MacOS Sierra 上安装 RocksDB：python-rocksdb 需要额外的编译器参数：

$ CFLAGS='-std=c++11 -stdlib=libc++ -mmacosx-version-min=10.10' \\
pip install -U –no-cache python-rocksdb

• 完整的 bundle 清单与依赖声明可以在 setup.py 的 BUNDLES 集合以及 requirements/extras/ 目录下找到（每个 .txt 对应一个扩展）。更多安装细节可参考 docs/userguide/installation.rst。

设计考量

Modern Python（现代 Python） Faust 使用当前 Python 3 的特性，如 async/await 与类型注解，并且代码库是静态类型化的，由 mypy 类型检查器校验。你在编写 Faust 应用时也可以使用类型注解，但这并非强制要求（见 examples/hello_world.py，其中未使用注解依然可以正常运行）。仓库还携带 py.typed 标记（见 faust/py.typed），符合 PEP 561 类型标注分发规范。

Library（库式设计） Faust 被设计为可嵌入任何现有 Python 程序的库，同时附带了一些辅助工具，让你无需大量样板代码即可完成部署。

Supervised（受监督运行） Faust worker 由众多服务（Service）按特定顺序启停组成，这些服务可以被监督器管理；但一旦遇到不可恢复的错误（例如无法恢复与 Kafka 断开的连接），Faust 的设计是直接崩溃（crash）。为此，Faust 被设计为运行在进程监督工具之内，例如 supervisord、Circus，或操作系统自带的进程管理机制。仓库 extra/supervisord/、extra/systemd/ 目录中就提供了现成的 supervisord 与 systemd 配置示例（如 extra/systemd/faust.service）。在 Agent 层面，源码 _new_supervisor() 也会为每个 Agent 建立监督策略（默认 max_restarts=100.0, over=1.0），单个 Actor 崩溃后会被自动替换重建（见 faust/agents/agent.py）。

Extensible（可扩展） Faust 抽象了存储（storage）、序列化器（serializer），甚至消息传输层（transport），开发者可以轻松为其增加新能力，并与现有系统集成。例如存储层有 faust/stores/（内存、RocksDB），序列化层有 faust/serializers/（含 registry 可注册自定义 codec），传输层有 faust/transport/drivers/（aiokafka 与 confluent 两种 Kafka 客户端驱动）。

Lean（精简） Faust 的源码短小易读，是学习 Kafka 流处理系统工作原理的良好起点。仓库入口模块 faust/init.py 的注释还贴心给出了源码阅读路线：faust/app.py（应用）→ faust/channels.py（通信通道）→ faust/topics.py（命名通道，如 Kafka topic）→ faust/streams.py（遍历通道事件的流）→ faust/tables.py（存储数据）→ faust/agents.py（组合以上全部）。

继续阅读

• 完整入门代码：examples/hello_world.py、examples/word_count.py
• 快速上手手册：docs/playbooks/quickstart.rst
• 核心概念详解：docs/userguide/agents.rst、docs/userguide/tables.rst、docs/userguide/streams.rst、docs/userguide/application.rst
• 常见问题：docs/faq.rst

一句话总结：Faust = 纯 Python 语法 + asyncio 异步模型 + Kafka 事件流 + 分布式有状态 Table。只要有一个 Kafka broker 和 Python 3.6+，你就能开始构建属于自己的分布式流处理应用。

赞

分享

• 流处理
• 消息队列
• 后端

【免费下载链接】faust

Python Stream Processing

项目地址：
https://gitcode.com/gh_mirrors/fa/faust

点击查看 免费下载

相关推荐

• Faust：用纯 Python 与 asyncio 构建的 Kafka 流处理与状态管理库

  Faust：用纯 Python 与 asyncio 构建的 Kafka 流处理与状态管理库 Faust 是一个将 Kafka Streams 的思想移植到 Py

  流处理
  消息队列
  后端

• FastStream Confluent KafkaBroker 入门指南：用 Confluent Kafka Python 客户端构建事件流应用

  FastStream Confluent KafkaBroker 入门指南：用 Confluent Kafka Python 客户端构建事件流应用 本篇指南围绕

  后端
  消息队列
  微服务

• 3步彻底解决Yuzu模拟器卡顿问题：2024终极优化完整指南

  3步彻底解决Yuzu模拟器卡顿问题：2024终极优化完整指南 还在为Yuzu模拟器的卡顿、闪退问题烦恼吗？作为专注模拟器优化的技术顾问，我深知这些问题的困扰。本

  游戏开发

上一篇：
PKHeX自动合法性插件完全攻略：告别手动调整个体值的烦恼时代

下一篇：
WritingTools隐私保护功能：如何使用本地LLM确保数据安全

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考