Faust-简洁高效的Python流处理库

在分布式系统和实时数据处理中，流处理是十分重要的技术。在数据密集型应用中，数据快速到达，转瞬即逝，需要及时进行处理，流式处理强调数据和事件的处理速度，对性能和可靠性有较高的要求。

创新互联建站自成立以来，一直致力于为企业提供从网站策划、网站设计、做网站、成都做网站、电子商务、网站推广、网站优化到为企业提供个性化软件开发等基于互联网的全面整合营销服务。公司拥有丰富的网站建设和互联网应用系统开发管理经验、成熟的应用系统解决方案、优秀的网站开发工程师团队及专业的网站设计师团队。

流处理框架包括：Storm，Spark Streaming 和 Flink 等，而 Kafka 也不甘示弱，推出了分布式流处理平台 Kafka Streams。 Faust 把 Kafka Streams 带到了 Python，并实现了抽象和优化，为数据和事件的流处理提供了一个高效便利的框架。

简介

Faust，是 robinhood 在 Github 上开源的 Python 流处理库，目前版本为 1.10.4。

Faust 把 Kafka Streams 的概念带到了 Python，提供了包括流处理和事件处理的模式。Faust 使用纯 Python 实现，使得开发者可以使用包括 NumPy, PyTorch, Pandas 等的库进行数据处理。

Faust 实现简洁优雅，使用简单，性能优秀，且具有高可用、分布式、灵活性高的特点。目前 Faust 已被用于构建高性能分布式系统和实时数据管道中。

Faust

使用

Faust 需求 Python 3.6 或以上，且需要可用的 Kafka >= 0.10 服务。使用 pip 安装：

 
 
 
 
  
  
  
  $ pip install -U faust

此外，一些额外的特性需要额外的依赖，如 rocksdb，可以用来作为 Faust 在生产环境中的存储，以及 Redis，可以在开启缓存时使用。

Faust

安装完成以后，就可以在项目中使用了。我们来看一个简单的例子：

 
 
 
 
  
  
  
  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 创建一个 Faust 应用，并配置应用的名字、Kafka broker 和序列化方式。

然后，我们创建一个主题，这跟 Kafka 中的主题是对应的。

Faust 利用 Python 3.6+ 的异步语法 async，定义异步函数 greet，并注册为 Faust 应用的一个 agent。函数接收实时的数据集合 greetings，并异步地对每项数据进行输出。

把上述代码保存为 hello_world.py，并在命令行启动工作者：

 
 
 
 
  
  
  
  $ faust -A hello_world worker -l info

该 Faust 工作者就会从 Kafka 中实时读取数据并处理。

我们可以发送一些数据来观察效果：

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

上述命令发送了一条消息，执行后，我们就能在工作者的命令行中看到这条消息。

Faust 还充分利用了 Python 的类型提示，能够方便地定义数据模型：

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

Faust

总结

Faust 把 Kafka Streams 带到了 Python 中，实现了简洁高效的数据流处理。其使用简单的装饰器和基于类型提示机的据模型，就能定义实现数据的处理逻辑;充分利用了 Python 的 async 异步机制，和其他高性能的异步库，实现了高效性能;其使用 Python 实现，使用开发者可以无缝对接其他数据处理和大数据相关功能。

本文标题：Faust-简洁高效的Python流处理库
分享地址：http://www.shufengxianlan.com/qtweb/news15/121465.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：创新互联

猜你还喜欢下面的内容