在分布式系統(tǒng)和實時數(shù)據(jù)處理中，流處理是十分重要的技術(shù)。在數(shù)據(jù)密集型應(yīng)用中，數(shù)據(jù)快速到達，轉(zhuǎn)瞬即逝，需要及時進行處理，流式處理強調(diào)數(shù)據(jù)和事件的處理速度，對性能和可靠性有較高的要求。

流處理框架包括：Storm，Spark Streaming 和 Flink 等，而 Kafka 也不甘示弱，推出了分布式流處理平臺 Kafka Streams。Faust 把 Kafka Streams 帶到了 Python，并實現(xiàn)了抽象和優(yōu)化，為數(shù)據(jù)和事件的流處理提供了一個高效便利的框架。

簡介

Faust，是 robinhood 在 Github 上開源的 Python 流處理庫，目前版本為 1.10.4。

Faust 把 Kafka Streams 的概念帶到了 Python，提供了包括流處理和事件處理的模式。Faust 使用純 Python 實現(xiàn)，使得開發(fā)者可以使用包括 NumPy, PyTorch, Pandas 等的庫進行數(shù)據(jù)處理。

Faust 實現(xiàn)簡潔優(yōu)雅，使用簡單，性能優(yōu)秀，且具有高可用、分布式、靈活性高的特點。目前 Faust 已被用于構(gòu)建高性能分布式系統(tǒng)和實時數(shù)據(jù)管道中。

使用

Faust 需求 Python 3.6 或以上，且需要可用的 Kafka >= 0.10 服務(wù)。使用 pip 安裝：

$ pip install -U faust

此外，一些額外的特性需要額外的依賴，如 rocksdb，可以用來作為 Faust 在生產(chǎn)環(huán)境中的存儲，以及 Redis，可以在開啟緩存時使用。

安裝完成以后，就可以在項目中使用了。我們來看一個簡單的例子：

import faust 
 
app = faust.App( 
    'hello-world', 
    broker='kafka://localhost:9092', 
    value_serializer='raw', 
) 
 
greetings_topic = app.topic('greetings') 
 
@app.agent(greetings_topic) 
async def greet(greetings): 
    async for greeting in greetings: 
        print(greeting) 

首先，我們使用 faust.App 創(chuàng)建一個 Faust 應(yīng)用，并配置應(yīng)用的名字、Kafka broker 和序列化方式。

然后，我們創(chuàng)建一個主題，這跟 Kafka 中的主題是對應(yīng)的。

Faust 利用 Python 3.6+ 的異步語法 async，定義異步函數(shù) greet，并注冊為 Faust 應(yīng)用的一個 agent。函數(shù)接收實時的數(shù)據(jù)集合 greetings，并異步地對每項數(shù)據(jù)進行輸出。

把上述代碼保存為 hello_world.py，并在命令行啟動工作者：

$ faust -A hello_world worker -l info

該 Faust 工作者就會從 Kafka 中實時讀取數(shù)據(jù)并處理。

我們可以發(fā)送一些數(shù)據(jù)來觀察效果：

$ faust -A hello_world send @greet "Hello Faust"

上述命令發(fā)送了一條消息，執(zhí)行后，我們就能在工作者的命令行中看到這條消息。

Faust 還充分利用了 Python 的類型提示，能夠方便地定義數(shù)據(jù)模型：

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() 

總結(jié)