Python入門學(xué)習(xí)之Python流處理過程
Faust是一個(gè)流處理庫,將kafka流中的思想移植到Python中
它被用于Robinhood去構(gòu)建高性能的分布式系統(tǒng)和實(shí)時(shí)數(shù)據(jù)通道,每天處理數(shù)十億的數(shù)據(jù)。
Faust同時(shí)提供流處理和事件處理同類型的工具分享例如:Kafka Streams, Apache Spark/Storm/Samza/Flink
它不需要使用一個(gè)DSL,僅需要用到Python!這意味著你在做流處理的時(shí)候可以使用所有你喜歡的Python庫:
NumPy, PyTorch, Pandas, NLTK, Django, Flask, SQLAlchemy等等。
由于需要使用新的async/await語法和變量類型注釋方法,F(xiàn)aust需要使用Python3.6以上的版本。
這里有一個(gè)處理輸入命令流的示例:
這個(gè)agent裝飾器定義了一個(gè)“流處理器”,它本質(zhì)上是一個(gè)Kafka topic,并且可以對(duì)接收到的每個(gè)事件做一些處理。
agent是一個(gè)async def的函數(shù)因此它還可以異步執(zhí)行其他操作如web請(qǐng)求。
這個(gè)系統(tǒng)可以持久化狀態(tài),執(zhí)行方式類似于數(shù)據(jù)庫。表被命名成分布式的key/value儲(chǔ)存,你可以使用常規(guī)的Python字典來做這件事。
在每臺(tái)機(jī)器上的本地用c++編寫的超快嵌入式數(shù)據(jù)庫(被稱為RocksDB)存儲(chǔ)表。
表還可以存儲(chǔ)可選的“窗口”聚合計(jì)數(shù),以便跟蹤“前一天的單擊次數(shù)”或“前一個(gè)小時(shí)的單擊次數(shù)”。與Kafka流一樣,我們支持滾動(dòng)、跳躍和滑動(dòng)時(shí)間窗口,舊窗口可以過期以阻止數(shù)據(jù)填充。
為了提高可靠性使用Kafka topic作為“預(yù)寫日志”
當(dāng)一個(gè)密鑰被更改時(shí),我們將其發(fā)布到更新的日志上。備用節(jié)點(diǎn)使用這個(gè)更新日志來保存數(shù)據(jù)的精確副本,并在任何節(jié)點(diǎn)發(fā)生故障時(shí)支持立即恢復(fù)。
對(duì)于用戶來說,表只是一個(gè)字典,但是數(shù)據(jù)在重新啟動(dòng)和跨節(jié)點(diǎn)復(fù)制之間存在,所以在故障發(fā)生時(shí)其他節(jié)點(diǎn)可以自動(dòng)接管。
您可以通過URL統(tǒng)計(jì)頁面瀏覽數(shù)量:
發(fā)送到Kafka topic的數(shù)據(jù)是分區(qū)的,這意味著點(diǎn)擊數(shù)將用URL的這種方式進(jìn)行分片。因此,同一個(gè)URL的每個(gè)計(jì)數(shù)都會(huì)立刻被傳遞給同一個(gè)Faust worker實(shí)例。
Faust支持任何類型的流數(shù)據(jù)字節(jié)、Unicode和序列化結(jié)構(gòu),同時(shí)也支持使用現(xiàn)代Python語法的“模型”來描述流中的keys和value是如何被序列化的。
Faust是靜態(tài)類型的,使用mypy類型檢查器,所以您在編寫應(yīng)用程序時(shí)可以充分利用靜態(tài)類型的優(yōu)勢(shì)。
Faust源代碼很小,組織良好,是學(xué)習(xí)Kafka流實(shí)現(xiàn)的好資源。
在引言頁學(xué)習(xí)更多關(guān)于Faust的知識(shí).jpg
去閱讀更多關(guān)于Faust,系統(tǒng)請(qǐng)求,安裝指導(dǎo),論壇資源等等,或者直接訪問快速開始的教程。在一個(gè)編寫流處理的應(yīng)用中去查看關(guān)于Faust應(yīng)用,然后通過使用者手冊(cè)深入探討。深層次的信息都根據(jù)不同主題在這個(gè)手冊(cè)中進(jìn)行說明
Faust簡(jiǎn)介
Faust非常容易使用。在學(xué)習(xí)其他的流處理方法時(shí),你總是需要從一個(gè)復(fù)雜的hello-world工程和相應(yīng)的基礎(chǔ)要求開始學(xué)習(xí)。Faust僅僅需要Kafka,剩下的就是只需要Python,如果你知道Python的話你就可以直接使用Faust去做流處理的工作了,并且它可以整合和他相關(guān)的一切。
這兒有一個(gè)簡(jiǎn)單的應(yīng)用程序你可以做:源代碼是Python的
您可能會(huì)被async和await這兩個(gè)關(guān)鍵字嚇到,但是您在使用Faust時(shí)不需要知道asyncio是如何工作的:只要模仿這些例子就可以得到您想要的結(jié)果。
示例應(yīng)用程序啟動(dòng)兩個(gè)任務(wù):一個(gè)是處理流,另一個(gè)是向流發(fā)送事件的后臺(tái)線程。在實(shí)際的應(yīng)用程序中,您的系統(tǒng)將向Kafka topic發(fā)布事件,您的處理器可以從Kafka topic獲取事件信息,并且只需要后臺(tái)線程將數(shù)據(jù)輸入到我們的示例中。
高可用性
Faust是高度可用的,并且可以在網(wǎng)絡(luò)問題和服務(wù)器崩潰中生存下來。在節(jié)點(diǎn)失敗的情況下,它可以自動(dòng)恢復(fù),并且表將接管備用節(jié)點(diǎn)。
分布式的
根據(jù)您的應(yīng)用程序的需要啟動(dòng)更多實(shí)例。
快速
一個(gè)單內(nèi)核的Faust worker實(shí)例已經(jīng)可以每秒處理數(shù)萬個(gè)事件,我們有理由相信,一旦我們能夠支持一個(gè)更優(yōu)化的Kafka客戶端,吞吐量就會(huì)增加。
靈活性
Faust就是Python,而流是一個(gè)無限的異步迭代器。如果您知道如何使用Python,那么您已經(jīng)知道如何使用Faust,它可以與您喜歡的Python庫一起使用,比如Django、Flask、SQLAlchemy、NTLK、NumPy、Scikit、TensorFlow等等。
安裝
您可以通過Python包或從源文件中安裝Faust
使用pip安裝它:
綁定
Faust還定義了一組setuptools擴(kuò)展,可以用來安裝Faust,并且有一個(gè)給定特性的依賴關(guān)系。
您可以在您的需求中或在pip命令行中使用方括號(hào)來指定它們。使用逗號(hào)分隔多個(gè)包:
以下的綁定均是有效的:
商店
最優(yōu)化
傳感器
事件循環(huán)
調(diào)試
下載并從源文件中安裝
您可以這樣安裝它:
如果當(dāng)前沒有使用virtualenv,則必須以特權(quán)用戶的身份執(zhí)行最后一個(gè)命令。
使用開發(fā)版本
您可以使用以下pip命令安裝Faust的最新版本:
常見問題
Faust可以在Django/Flask/etc上使用嗎?
使用gevent
這種方法適用于任何可以與gevent一起工作的阻塞Python庫。
使用gevent需要您安裝aiogevent模塊,您可以將其作為Faust的包進(jìn)行安裝:
然后要真正的使用gevent作為事件循環(huán),您要么在faust程序中使用-L <faust --loop>
命令:
要么在你腳本的前面加入import mode.loop.gevent
警告
非常重要的是,它位于模塊的最頂端,并且在導(dǎo)入庫之前執(zhí)行。
Faust可以在Tornado上使用嗎?
可以!使用tornado.platform.asyncio
鏈接:http://www.tornadoweb.org/en/stable/asyncio.html
Faust可以在Twisted上使用嗎?
可以!使用asyncio反應(yīng)器實(shí)現(xiàn):
鏈接:https://twistedmatrix.com/documents/17.1.0/api/twisted.internet.asyncioreactor.html
是否支持Python3.5或者更早的版本?
目前還沒有支持Python 3.5的計(jì)劃,但是歡迎您為這個(gè)項(xiàng)目做出貢獻(xiàn)。
以下是實(shí)現(xiàn)這一目標(biāo)所需的一些步驟
- 源代碼轉(zhuǎn)換以重寫變量注釋到注釋
示例,代碼:
- 重寫異步函數(shù)的源代碼轉(zhuǎn)換
示例,代碼:
必須重寫:
你將支持Python2嗎?
目前還沒有支持Python 2的計(jì)劃,但是歡迎您為項(xiàng)目做貢獻(xiàn)(上面問題中的細(xì)節(jié)也與Python 2相關(guān))。
在本地運(yùn)行Faust應(yīng)用程序時(shí),我得到的打開文件的最大數(shù)量超過了RocksDB的錯(cuò)誤。我該怎么解決這個(gè)問題呢
您可能需要增加打開文件的最大數(shù)量的限制。
資源
編碼規(guī)范
在項(xiàng)目代碼庫、問題跟蹤器、聊天室和郵件列表中進(jìn)行交互的每個(gè)人都應(yīng)該遵循《Faust行為準(zhǔn)則》。
作為這些項(xiàng)目的貢獻(xiàn)者和維護(hù)者,為了培養(yǎng)開放和受歡迎的社區(qū),我們承諾尊重所有通過報(bào)告問題、發(fā)布特性請(qǐng)求、更新文檔、提交合并請(qǐng)求或補(bǔ)丁和其他活動(dòng)的人。
我們致力于使參與這些項(xiàng)目的每個(gè)人都無騷擾體驗(yàn),不論其經(jīng)驗(yàn)水平、性別、性別認(rèn)同和表現(xiàn)、性取向、殘疾、個(gè)人外貌、體型、種族、種族、年齡、宗教或國籍。
參與者不良行為包括:
- 性化的語言或意象的使用
- 個(gè)人人身攻擊
- 惡意破壞或侮辱/侮辱性的評(píng)論
- 公共或者私人的騷擾
- 未經(jīng)明確許可,發(fā)布他人的私人信息,如住址或電子地址
- 其他不道德或不專業(yè)的行為。
項(xiàng)目維護(hù)人員有權(quán)利和責(zé)任刪除、編輯或拒絕評(píng)論、提交、代碼、wiki編輯、問題和其他與行為準(zhǔn)則不一致的貢獻(xiàn)。通過采用這一行為準(zhǔn)則,項(xiàng)目維護(hù)者承諾在管理這個(gè)項(xiàng)目的每個(gè)方面都公平、一致地應(yīng)用這些原則。不遵守或執(zhí)行行為準(zhǔn)則的項(xiàng)目維護(hù)者可能被永久地從項(xiàng)目團(tuán)隊(duì)中刪除。
當(dāng)個(gè)人代表項(xiàng)目或社區(qū)時(shí),此行為準(zhǔn)則適用于項(xiàng)目空間和公共空間。
可以通過創(chuàng)建一個(gè)問題或聯(lián)系一個(gè)或多個(gè)項(xiàng)目負(fù)責(zé)人來舉報(bào)虐待、騷擾或其他不可接受行為。
以上就是Python入門學(xué)習(xí)Python流處理過程的詳細(xì)內(nèi)容,更多關(guān)于Python入門的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
django orm 通過related_name反向查詢的方法
今天小編就為大家分享一篇django orm 通過related_name反向查詢的方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2018-12-12django 實(shí)現(xiàn)電子支付功能的示例代碼
這篇文章主要介紹了django 實(shí)現(xiàn)電子支付功能的示例代碼,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧2018-07-07Django框架orM與自定義SQL語句混合事務(wù)控制操作
這篇文章主要介紹了Django框架orM與自定義SQL語句混合事務(wù)控制操作,結(jié)合實(shí)例形式分析了同一個(gè)方法里面既有ORM又有自定義SQL 語句的情況下事務(wù)控制相關(guān)操作技巧,需要的朋友可以參考下2019-06-06Python多線程threading join和守護(hù)線程setDeamon原理詳解
這篇文章主要介紹了Python多線程threading join和守護(hù)線程setDeamon原理詳解,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-03-03Python基礎(chǔ)教程之if判斷,while循環(huán),循環(huán)嵌套
這篇文章主要介紹了Python基礎(chǔ)教程之if判斷,while循環(huán),循環(huán)嵌套 的相關(guān)知識(shí),非常不錯(cuò),具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-04-04pandas讀取csv格式數(shù)據(jù)時(shí)header參數(shù)設(shè)置方法
本文主要介紹了pandas讀取csv格式數(shù)據(jù)時(shí)header參數(shù)設(shè)置方法,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2022-02-02Python實(shí)現(xiàn)隨機(jī)選擇元素功能
這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)隨機(jī)選擇元素功能,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-09-09