Flink SQL 在网易云音乐的产品化实践
摘要:本文由網(wǎng)易云音樂數(shù)據(jù)智能部資深數(shù)據(jù)平臺開發(fā)工程師蔣文偉分享,主要介紹 Flink SQL 在云音樂的產(chǎn)品化實踐。分享內(nèi)容如下:
一、背景簡介
1.Flink in Music
先簡單的介紹下云音樂的現(xiàn)狀,目前音樂這邊的客戶端日志,服務(wù)端日志大概在每日大千億條左右,維度表數(shù)據(jù)源像 Redis,MySQL 這些大概有上百個。而服務(wù)的實時計算任務(wù)開發(fā)的人員有上百名,其中不僅包擴數(shù)據(jù)開發(fā)工程師,分析師,也包括算法,后臺業(yè)務(wù)等同學。這些同學們也累積開發(fā)了上千個實時計算任務(wù),這些任務(wù)不僅有統(tǒng)計任務(wù),還有些一線業(yè)務(wù),比如排行榜,實時熱度等。
2.應(yīng)用場景
這里我稍微列舉了一些業(yè)務(wù)場景,比如我們的內(nèi)容分發(fā)、實時數(shù)倉、算法推薦,還有索引任務(wù)、實時監(jiān)控,AB test 等,幾乎涵蓋了整個數(shù)據(jù)鏈路中的大部分業(yè)務(wù)場景,可以說我們現(xiàn)在的業(yè)務(wù)已經(jīng)離不開 Flink 的體系,那后面我們會以其中幾個場景為例,看看我們在這些場景中使用原生的 Flink 會遇到那些問題。
■ 內(nèi)容分發(fā)
第一個介紹的場景是分發(fā),分發(fā)是個非常典型的場景。它會根據(jù)一定條件對數(shù)據(jù)流進行劃分,把輸入數(shù)據(jù)流切分成多個子流。
一般情況下分發(fā)會是整個數(shù)據(jù)鏈路的上游,所以相對來說這類任務(wù)非常重要,那么我們在這個場景中會遇到什么問題呢?
- 問題1:開發(fā)效率低,業(yè)務(wù)標準流程難以復(fù)用
首先是開發(fā)效率低下,這類業(yè)務(wù)邏輯非常簡單,核心開發(fā)工作其實就是個 where 篩選,但是傳統(tǒng)的開發(fā)方式需要用戶了解很多額外的東西,比如 HDFS 的定時清理功能,如果這個組件交由用戶開發(fā),勢必要開放權(quán)限,那么就可能會導(dǎo)致 HDFS 倉庫文件被誤刪等安全事故,所以我們需要建立一套統(tǒng)一框架,且可以提供一系列標準化的組件,用戶僅需要關(guān)心其核心業(yè)務(wù)邏輯即可。
- 問題2:學習成本高
第二個問題是學習成本較高,SQL 是一種非常優(yōu)秀的數(shù)據(jù)處理語言,很多同學也都會,但是 Flink SQL 的配置卻沒普通 SQL 那么簡單。Flink SQL 要求用戶對每個組件的配置都非常熟悉,這是一個 HDFS 的 sink 操作,需要在 SQL 中配置輸出目錄,分區(qū)字段,文件大小,keytab,壓縮格式等一系列的參數(shù),而這些參數(shù)需要用戶通過文檔來學習。
針對這個,我們需要一種前端上的優(yōu)化,通過給用戶可視化的提示,來簡化配置,降低學習成本。
- 問題3:外部環(huán)境混亂
第三,對一個穩(wěn)定性,以及性能要求比較高的任務(wù)來說,所有的這些監(jiān)控、報警的配套體系建設(shè)也都是必不可少的。
■ 特征快照
第二個例子,特征快照,先簡單的說下什么是特征快照,特征快照簡單的可以理解成把特征的歷史版本進行存儲,為什么要這么做呢,因為特征是動態(tài)變化的,每個事件未必能保準順序到達,一個用戶先點喜歡 DJ,在播放歌曲,再點擊喜歡了動漫,我們最終這次播放關(guān)聯(lián)的應(yīng)該是 DJ 而不是動漫,但按照時間序可能就是錯的,所以,對一每個版本的特征和 tag,都會有其唯一的 traceid 來進行管理,也就是一個 traceid 一個特征版本,這塊在實時機器學習場景使用的非常廣泛。
這邊可以看到任務(wù)的流程圖,包括數(shù)據(jù)清洗,收集,抽樣,去重,join 等流程,這些流程也有很多業(yè)務(wù)邏輯,像 join 這個流程如果 join 不上怎么辦,是放在內(nèi)存里等,還是再次回流到 kafka 等待下一輪匹配,亦或是使用降級方案。相對來說,對穩(wěn)定性、兜底方案等都有較多要求。
我們來看特征快照場景下會遇到哪些問題。
- 第一,開發(fā)成本較高,調(diào)試也比較復(fù)雜。上述提到我們有很多種模塊共同組成這個任務(wù),如果通過傳統(tǒng)的日志打印,很難進行調(diào)試和維護。
- 第二,特殊的需求沒有辦法被快速的滿足。像抽樣功能完全取決于業(yè)務(wù),比較難以抽象和復(fù)用。
其它的一些問題,包括調(diào)優(yōu)問題、運維問題、監(jiān)控的問題,我歸納為技術(shù)賦能。這些問題不能推給用戶來做,只能由平臺來進行統(tǒng)一的管理和維護。
綜上所述,我們的產(chǎn)品化目標也出來了,我們希望打造一套能降低用戶學習成本、運維成本,提升開發(fā)效率,并在全方位進行一個賦能的一站式的實時計算平臺。也就是我們的主要介紹的產(chǎn)品,音樂的實時計算 Notebook 服務(wù)。
二、云音樂的實時計算 Notebook 服務(wù)
1.NoteBook with block
我們的 notebook 服務(wù)是由多個 block 快組成的,每個 block 可以自由組合,比如 SQL 類型的塊之后加 2 個 sink 的組件,再加一個 source 組件,都可以。
同樣,每個 block 塊擁有子類型,也就是二級類型,比如一個 sink,會有一些類似 MySQL sink,Redis sink 的子類型,他們都以插件的形式進行加載,因此主類型可以被快速擴展,子類型也同樣可以被快速擴展,組件開發(fā)就變得非常方便。
而整個 notebook 執(zhí)行的過程,也會按照順序由上至下的執(zhí)行每一個 block,同時我們的 block 中除了 sink 之外都支持在頁面的 debug 服務(wù)。
詳細分享一下目前支持的 block 類型。
- 第一種就是 SQL block,支持 Flink SQL,當然我們也在中間做了一些優(yōu)化。比如建立自己的 catalog 以及 function 管理中心,自定義的一些參數(shù) set 語法等。
- 第二種是 custom block,我們會提供給用戶一套標準的 API,通過 API 就可以進行個性化的開發(fā),類似一個有限條件下的 Flink 的 jar 包任務(wù)。
source 和 sink 類型,他們比純 SQL 類型有更多的優(yōu)勢,比如同樣的功能在 SQL 中也可以實現(xiàn),但是需要加很多 set config 配置。而現(xiàn)在,不僅 catalog DB 支持搜索,并且有些配置也將更加直觀的被呈現(xiàn)。就像 HDFS sink,他的歸檔大小,序列化類型,前綴,等等常規(guī)的配置就直接可以讓用戶進行選擇。
第二個優(yōu)勢,我們可以通過可視化 block 來提供組件,比如 source 端的動態(tài)限流,可以開啟一個 source 的 slot 來做全局 config 的輪訓,比如 sink 有類似 HDFS 文件過期清理或者小文件合并的功能。類似的標準組件可以通過 block 塊更直觀的提供給用戶。
整個平臺的頁面大概介紹到這里,這里是些小的歸納總結(jié),良好的交互也就是跟簡單直觀的操作我們剛剛介紹了一些,良好的擴展,也就是我們的 block 的插件化,元數(shù)據(jù)中心以及監(jiān)控配套等,我們先看如何用這套 notebook 來解決實際例子。
■ Snapshot 場景的應(yīng)用
首先是 snapshot 場景的應(yīng)用,我們可以看到這個場景中,鏈路上的 4 種任務(wù)分別通過不同的 block 類型實現(xiàn)。
- Clollect,通過 SQL 就能完成;
- ETL,會有一些業(yè)務(wù)上的抽樣方法和去重邏輯,所以會通過自定義 source 來實現(xiàn);
- Snapshut join,除了 join 操作之外,可能會有業(yè)務(wù)降級方案等一系列非標需求,通過 SQL 無法完成,需要通過自定義的 transform 組件來實現(xiàn) ;
- Extrect,通過可視化 sink來進行最后樣本文件的落盤。
2.功能實現(xiàn)
說完前端功能,我們再來說下實現(xiàn)。
整套 block notebook 在執(zhí)行的時候,都會找到對應(yīng)類型的 Interpreter,每個 Interpreter 再通過子類型去發(fā)現(xiàn)真正的實現(xiàn)類。像第一個 source 指向 MySQL 的 blog 日志訂閱,最終會創(chuàng)建一個 MySQL 的 blog 日志訂閱 source 實現(xiàn),每個 Interpreter 都會接收到所有上游執(zhí)行過的所有結(jié)果,這個結(jié)果統(tǒng)一以 table 的方式進行流轉(zhuǎn),在沒有指定的情況下,默認使用離自己最近的一個 table 作為輸入。
■ Block Structure
這是 notebook 的數(shù)據(jù)結(jié)構(gòu),屬于 JobContext。作為一個全局的共享的內(nèi)容,里面包含著一個可執(zhí)行環(huán)境和一些全局共享的配置。blockList 數(shù)組里每一個都是加載一個 interpreter 具體事件類。如果業(yè)務(wù)上有需求,擴展一個 block 也非常簡單,只要實現(xiàn)一個接口即可,其他的部分都交給框架來完成。
■ 提交執(zhí)行
以上是 block 的內(nèi)部執(zhí)行邏輯,我們再來單獨講一下提交的服務(wù)。
首先,服務(wù)會判斷任務(wù)中的所用到的插件和依賴的文件,然后通過一些機制來確定主程序版本,這么做的原因很簡單,我們經(jīng)常會做升級,而升級之后可能執(zhí)行計劃可能會變動,有些用戶停止任務(wù)后再通過 checkpoint 可能就起不來了,通過這個服務(wù)來實現(xiàn)多版本就非常簡單了。我們還可以根據(jù)一些條件,比如任務(wù)是否是 checkpoint 啟動的等來判斷主程序版本。最后提交服務(wù)會通過一些邏輯對集群和隊列的進行智能選擇。
■ 整體架構(gòu)
通過以上對整體架構(gòu)的分享,想必大家也就比較清晰了。上層會有一層 notebook 的 server 的服務(wù),下層會有一個 submit 的提交服務(wù)和一個 debug server。外圍會有元數(shù)據(jù)管理中心來管理我們的 catalog,然后還有權(quán)限或者一些其他的外部系統(tǒng)來進行輔助。
整個體系最大的好處在于,用戶的代碼都在框架的范圍內(nèi)進行執(zhí)行,這樣的話我們可以快速的進行性能優(yōu)化和功能調(diào)整。
三、性能優(yōu)化
團隊對整個平臺有非常多的性能優(yōu)化,這里只能拋磚引玉的說幾點。
1.Table source 優(yōu)化插件
第一個就是 Table source 的優(yōu)化。生產(chǎn)過程中的遇到的性能問題,如:
- 原生的 KafkaTableSource 由于需要解析 Schema 無法將反序列化過程提取出來,而 Kafka 的并發(fā)受制于業(yè)務(wù)的 partition 數(shù)量,如果反序列化計算量要求較大,會造成性能瓶頸。
- 維表 Join 如果先進行 keyBy,可以提升緩存命中率。
我們的解決方案包括 3 個大的步驟。
1、動態(tài)代理方式獲取 byte 流。
2、將 byte 流進行二次處理:
- 動態(tài)的加載所需執(zhí)行的插件。
- 分配插件到 Map 和 Filter 算子 中去(保證插件變動時 ckp 啟動)。
- 調(diào)整并發(fā)。
- 反序列化成目標 Row。
3、返回 ProxyTableSource。這樣就可以無限的擴展功能。
這是一個優(yōu)化實例:降低由于序列化計算量過大而導(dǎo)致的 source 端性能瓶頸我們把有 5 個 kafka partition 的一個任務(wù)拆分成 5 個 source 跟 10 個 Deserialization 的并發(fā),從而提升了整個任務(wù)的吞吐。
2.多 Sink 合并
第二個是多 sink 合并,這塊高版本已經(jīng)有了,因為我們還在使用 1.10,所以才需要優(yōu)化,這塊就不詳細說了。
3.流量降級方案
第三個是我們的流量降級方案,這塊優(yōu)化涉及到一些體系的建設(shè)。一般來說,我們的分發(fā)任務(wù)會有 2 個 sink,會分發(fā)到主消息隊列和備份消息隊列,這 2 個隊列的 catalog 完全一致,只會通過 tag 進行區(qū)分,所以相關(guān)代碼開發(fā)也完全一致,下游讀取的時候,會根據(jù)任務(wù)的 tag 分析說我們讀主還是備。舉例,歸檔任務(wù),測試任務(wù),等會從備份流來讀取,這樣我們就對 kafka 集群進行了壓力分流。
四、運維監(jiān)控增強
介紹完性能優(yōu)化,我們來聊下運維和監(jiān)控。Flink 自己已經(jīng)帶有不少監(jiān)控,比如failover 次數(shù)、QPS、checkpoint 成功率,但是在生產(chǎn)過程中,這可能還不夠,所以會增加很多內(nèi)置監(jiān)控,并且對血緣進行收集。
此外我們做了 4 個智能診斷功能,包括內(nèi)存診斷、性能診斷、checkpoint 診斷和日志診斷,從而來判斷我們的任務(wù)運行情況。
這是我們之前看過架構(gòu)圖,紅色就是監(jiān)控診斷部分,由一個統(tǒng)一的數(shù)據(jù)收集服務(wù)來獲取信息,并提供給監(jiān)控報警等服務(wù)使用。
監(jiān)控增強這塊其實就是在在框架中增加很多內(nèi)置的指標,比如 sink 平均寫入時間,序列化錯誤率,維表 join 命中率,sink 寫入時間等等一系列的指標,這些參數(shù)也有助于我們更多維度的了解任務(wù)實際情況,當然我們可以選擇是否開啟這些監(jiān)控,以保證高峰期任務(wù)性能。
第二塊是血緣,它是數(shù)據(jù)治理中非常重要的一塊。我們收集數(shù)據(jù)血緣比較簡單,通過Block 的參數(shù)進行一個靜態(tài)解析來看所有的輸入源和輸出表,然后上報給數(shù)據(jù)中心,最終組成一個血緣的關(guān)系圖。
我們的智能診斷其實是利用所有監(jiān)控數(shù)據(jù)來綜合判斷一個任務(wù)是不是處于正常狀態(tài),包括四塊。
- 第一,內(nèi)存診斷
- 第二,性能診斷
- 第三,Checkpoint 診斷
- 第四,日志診斷,舉例說明,內(nèi)存診斷,它主要通過 GC 如何觸發(fā)報警,會有很多條判斷標準,比如 young gc 頻率,full gc 頻率, gc 耗時等等是否大于或者小于一定值,通過這些條件綜合來判斷一個任務(wù)是否存在 gc 異常,以及異常的等級。而日志會通過一定條件對錯誤日志進行篩選,這個條件可以是用戶自定義的,比如統(tǒng)計某個具體的 exception 的次數(shù),當達到統(tǒng)計次數(shù)時候進行報警的觸發(fā)。這樣的話,某些業(yè)務(wù)不想被 failover,又想收到異常報警,就通過簡單的 try catch 就能完成。
這是我們智能診斷的一個內(nèi)存診斷的前端的頁面,展示了一個實際任務(wù)的內(nèi)存的使用情況分布。
五、未來規(guī)劃
最后講一下對未來的一些規(guī)劃。
第一,體系建設(shè):權(quán)限體系將更加靈活,數(shù)據(jù)安全/用戶管理等配套將更新完善。
第二,Notebook 體系將引入其他數(shù)據(jù)計算引擎(如 druid 等)。
第三,擁抱新版本,現(xiàn)在的 Flink 1.12 有非常多的新特性,有些也非常吸引人。我們希望能積極的跟上社區(qū)版本。
活動推薦:
僅需99元即可體驗阿里云基于 Apache Flink 構(gòu)建的企業(yè)級產(chǎn)品-實時計算 Flink 版!點擊下方鏈接了解活動詳情:https://www.aliyun.com/product/bigdata/sc?utm_content=g_1000250506
原文鏈接:https://developer.aliyun.com/article/782796?
版權(quán)聲明:本文內(nèi)容由阿里云實名注冊用戶自發(fā)貢獻,版權(quán)歸原作者所有,阿里云開發(fā)者社區(qū)不擁有其著作權(quán),亦不承擔相應(yīng)法律責任。具體規(guī)則請查看《阿里云開發(fā)者社區(qū)用戶服務(wù)協(xié)議》和《阿里云開發(fā)者社區(qū)知識產(chǎn)權(quán)保護指引》。如果您發(fā)現(xiàn)本社區(qū)中有涉嫌抄襲的內(nèi)容,填寫侵權(quán)投訴表單進行舉報,一經(jīng)查實,本社區(qū)將立刻刪除涉嫌侵權(quán)內(nèi)容。總結(jié)
以上是生活随笔為你收集整理的Flink SQL 在网易云音乐的产品化实践的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 一文教你快速上手PyFlink
- 下一篇: 如何利用DataWorks OpenAP