Python入门学习Python流处理过程

目录

• Faust是一个流处理库
• 将kafka流中的思想移植到python中
• agent是一个async def的函数，因此它还可以异步执行其他操作
• 使用Kafka topic作为“预写日志”
• Faust支持任何类型的流数据
• Faust是静态类型的
• Faust简介
• 高可用性
• 分布式的
• 快速
• 灵活性
• 安装
• 绑定
• 下载并从源文件中安装
• 使用开发版本
• 常见问题

Faust是一个流处理库

将kafka流中的思想移植到Python中

它被用于Robinhood去构建高性能的分布式系统和实时数据通道，每天处理数十亿的数据。

Faust同时提供流处理和事件处理同类型的工具分享例如：Kafka Streams, Apache Spark/Storm/Samza/Flink

它不需要使用一个DSL，仅需要用到Python！这意味着你在做流处理的时候可以使用所有你喜欢的Python库：

NumPy, PyTorch, Pandas, NLTK, Django, Flask, SQLAlchemy等等。

由于需要使用新的async/await语法和变量类型注释方法，Faust需要使用Python3.6以上的版本。

这里有一个处http://www.cppcns.com理输入命令流的示例：

这个agent装饰器定义了一个“流处理器”，它本质上是一个Kafka topic，并且可以对接收到的每个事件做一些处理。

agent是一个async def的函数，因此它还可以异步执行其他操作

如web请求。

这个系统可以持久化状态，执行方式类似于数据库。表被命名成分布式的key/value储存，你可以使用常规的Python字典来做这件事。

在每台机器上的本地用c++编写的超快嵌入式数据库(被称为RocksDB)存储表。

表还可以存储可选的“窗口”聚合计数，以便跟踪“前一天的单击次数”或“前一个小时的单击次数”。与Kafka流一样，我们支持滚动、跳跃和滑动时间窗口，旧窗口可以过期以阻止数据填充。

为了提高可靠性

使用Kafka topic作编程客栈为“预写日志”

当一个密钥被更改时，我们将其发布到更新的日志上。备用节点使用这个更新日志来保存数据的精确副本，并在任何节点发生故障时支持立即恢复。

对于用户来说，表只是一个字典，但是数据在重新启动和跨节点复制之间存在，所以在故障发生时其他节点可以自动接管。

您可以通过URL统计页面浏览数量：

发送到Kafka topic的数据是分区的，这意味着点击数将用URL的这种方式进行分片。因此，同一个URL的每个计数都会立刻被传递给同一个Faust worker实例。

Faust支持任何类型的流数据

字节、Unicode和序列化结构，同时也支持使用现代Python语法的“模型”来描述流中的keys和value是如何被序列化的。

Faust是静态类型的

使用mypy类型检查器，所以您在编写应用程序时可以充分利用静态类型的优势。

Faust源代码很小，组织良好，是学习Kafka流实现的好资源。

在引言页学习更多关于Faust的知识.jpg

去阅读更多关于Faust，系统请求，安装指导，论坛资源等等，或者直接访问快速开始的教程。在一个编写流处理的应用中去查看关于Faust应用，然后通过使用者手册深入探讨。深层次的信息都根据不同主题在这个手册中进行说明

Faust简介

Faust非常容易使用。在学习其他的流处理方法时，你总是需要从一个复杂的hello-world工程和相应的基础要求开始学习。Faust仅仅需要Kafka，剩下的就是只需要Python，如果你知道Python的话你就可以直接使用Faust去做流处理的工作了，并且它可以整合和他相关的一切。

这儿有一个简单的应用程序你可以做：源代码是Python的

您可能会被async和await这两个关键字吓到，但是您在使用Faust时不需要知道asyncio是如何工作的：只要模仿这些例子就可以得到您想要的结果。

示例应用程序启动两个任务：一个是处理流，另一个是向流发送事件的后台线程。在实际的应用程序中，您的系统将向Kafka topic发布事件，您的处理器可以从Kafka topic获取事件信息，并且只需要后台线程将数据输入到我们的示例中。

高可用性

Faust是高度可用的，并且可以在网络问题和服务器崩溃中生存下来。在节点失败的情况下，它可以自动恢复，并且表将接管备用节点。

分布式的

根据您的应用程序的需要启动更多实例。

快速

一个单内核的Faust worker实例已经可以每秒处理数万个事件，我们有理由相信，一旦我们能够支持一个更优化的Kafka客户端，吞吐量就会增加。

灵活性

Faust就是Python，而流是一个无限的异步迭代器。如果您知道如何使用Python，那么您已经知道如何使用Faust，它可以与您喜欢的Python库一起使用，比如Django、Flask、SQLAlchemy、NTLK、NumPy、Scikit、TensorFwww.cppcns.comlow等等。

安装

您可以通过Python包或从源文件中安装Faust

使用pip安装它：

绑定

Faust还定义了一组setuptools扩展，可以用来安装Faust，并且有一个给定特性的依赖关系。

您www.cppcns.com可以在您的需求中或在pip命令行中使用方括号来指定它们。使用逗号分隔多个包：

以下的绑定均是有效的：

商店

最优化

传感器

事件循环

调试

下载并从源文件中安装

您可以这样安装它：

如果当前没有使用virtualenv，则必须以特权用户的身份执行最后一个命令。

使用开发版本

您可以使用以下pip命令安装Faust的最新版本：

常见问题

Faust可以在Django/Flask/etc上使用吗？

使用gevent

这种方法适用于任何可以与gevent一起工作的阻塞Python库。

使用gevent需要您安装aiogevent模块，您可以将其作为Faust的包进行安装：

然后要真正的使用gevent作为事件循环，您要么在faust程序中使用-L <faust --loop>

命令：

要么在你脚本的前面加入import mode.loop.gevent

记住：非常重要的一点是，它位于模块的最顶端，并且在导入库之前执行。

使用eventlet

这种方法适用于任何可以使用eventlet的阻塞Python库。

使用eventlet需要您安装aioeventlet模块，您可以将其安装为与Faust一起的捆绑包。

然后要实际使用eventlet作为事件循环，您要么在faust程序中使用-L <faust --loop>

命令：

要么在你脚本的前面加入import mode.loop.gevent

警告

非常重要的是，它位于模块的最顶端，并且在导入库之前执行。

Faust可以在ToyRjpNhwgrnado上使用吗？

可以！使用tornado.platform.asyncio

链接：http://www.tornadoweb.org/en/stable/asyncio.html

Faust可以在Twisted上使用吗？

可以！使用asyncio反应器实现：

链接：https://twistedmatrix.com/documents/17.1.0/api/twisted.internet.asyncioreactor.html

是否支持Python3.5或者更早的版本?

目前还没有支持Python 3.5的计划，但是欢迎您为这个项目做出贡献。

以下是实现这一目标所需的一些步骤

• 源代码转换以重写变量注释到注释

示例，代码：

• 重写异步函数的源代码转换

示例，代码：

必须重写：

你将支持Python2吗？

目前还没有支持Python 2的计划，但是欢迎您为项目做贡献(上面问题中的细节也与Python 2相关)。

在本地运行Faust应用程序时，我得到的打开文件的最大数量超过了RocksDB的错误。我该怎么解决这个问题呢

您可能需要增加打开文件的最大数量的限制。

资源

编码规范

在项目代码库、问题跟踪器、聊天室和邮件列表中进行交互的每个人都应该遵循《Faust行为准则》。

作为这些项目的贡献者和维护者，为了培养开放和受欢迎的社区，我们承诺尊重所有通过报告问题、发布特性请求、更新文档、提交合并请求或补丁和其他活动的人。

我们致力于使参与这些项目的每个人都无骚扰体验，不论其经验水平、性别、性别认同和表现、性取向、残疾、个人外貌、体型、种族、种族、年龄、宗教或国籍。

参与者不良行为包括:

• 性化的语言或意象的使用
• 个人人身攻击
• 恶意破坏或侮辱/侮辱性的评论
• 公共或者私人的骚扰
• 未经明确许可，发布他人的私人信息，如住址或电子地址
• 其他不道德或不专业的行为。

项目维护人员有权利和责任删除、编辑或拒绝评论、提交、代码、wiki编辑、问题和其他与行为准则不一致的贡献。通过采用这一行为准则，项目维护者承诺在管理这个项目的每个方面都公平、一致地应用这些原则。不遵守或执行行为准则的项目维护者可能被永久地从项目团队中删除。

当个人代表项目或社区时，此行为准则适用于项目空间和公共空间。

可以通过创建一个问题或联系一个或多个项目负责人来举报虐待、骚扰或其他不可接受行为。

以上就是Python入门学习Python流处理过程的详细内容，更多关于Python入门的资料请关注我们其它相关文章！