轉載自公衆号網易有數，網易雲音樂從2018年開始搭建實時計算平台，經過幾年的發展已經滲透到雲音樂的各個業務當中。本文是大愚老師的一篇實踐分享，将從一個日常運維問題出發，帶領大家了解雲音樂實時計算平台的一些工作進展和未來規劃。主要内容為：
- 平台功能
- 批流一體
- 未來規劃
GitHub 位址
https://github.com/apache/flink歡迎大家給 Flink 點贊送 star~
網易雲音樂實時數倉平台上線以後，經過一年半的發展，整體實時數倉已經初具規模，我們已有實時數倉表300+，運作中的任務數有1200+。其中1000左右的任務是SQL任務， Kafka總出口流量達到到18GB/S，總使用者數達到了200+。
資料量和使用者的增長也給資料平台的易用性以及穩定性帶來了了越來越多的挑戰，包含Kafka的穩定性、叢集的穩定性、運維工作的挑戰以及很多早期的技術債；業務的增長，暴露出了基建的薄弱，也給我們積累了很多平台建設和運維的經驗。
一、平台功能
我們平台整體的的功能大家可以參考《雲音樂實時數倉技術改造以及未來的一些規劃》，這裡将主要介紹我們最新的一些工作：
“我的任務延遲了，怎麼擴容都不行，這是為什麼？”
在日常運維工作中這是我們經常遇到的問題，往往也是比較耗費時間的問題。導緻這種這種問題的原因有很多，為了解決這個問題，我們做了一些工作來增強我們的運維能力。
1. IO名額完善
IO問題是導緻以上問題經常出現的原因之一，包含消息讀取效率、維表JOIN效率、SINK效率等等，第三方存儲的性能以及穩定性，直接影響實時任務的穩定性，為了快速定位相關問題，我們添加了很多IO相關Metric名額。
1.1 Kafka消費側的一些性能名額
1.2 讀取反序列化名額
包含：
- 反序列化的RT
- 反序列化的錯誤比例
在Format側我們開發了一套Format代理，支援在不修改原有format代碼的情況下，上報相關metirc名額，忽略錯誤資料等功能。隻要添加屬性format.proxy指定代理類就可以支援不同方式的Format封裝。
比如我們指定format.proxy=magina，就可以支援上報上述的性能名額；指定format.proxy=ds 就可以支援解析ds封裝的日志格式，使用被代理的Format解析DS中的Body部分，不需要單獨為DS封裝的日志格式開發Format，且同樣會上報性能相關名額，支援忽略錯誤消息等功能。
1.3 維表JOIN相關名額
在維表JOIN側, 我們添加了：
- 資料查詢的響應時間
- 本地緩存的命中率
- 查詢發生重試的比例
- 成功JOIN上的資料的比例等
1.4 資料寫入的一些性能名額
- 資料序列化的RT
- 資料寫入外部資料源的平均響應時間等
整套IO相關名額的實作，我們全部是在Flink Connector的頂層接口做了一些公共的封裝，重構了相關Connector的代碼，隻要按照我們自己的接口實作Connector，無需關心細節名額的上報，這些名額都會自動化的上報出來。
2. Kafka分區問題
Kafka分區的限制也是經常導緻我們程式性能無法擴充的原因，出于Exactly Once的實作、讀取性能、以及讀取穩定性的考慮，Flink采用主動拉取的方式讀取Kafka消息，這種方式限制了我們讀取Kafka消息的任務數，大大限制我們任務性能的擴張能力，以下面這個case為例：
SET 'table.exec.state.ttl' = '1h';
SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '10s';
SET 'table.exec.mini-batch.size' = '100000';
INSERT INTO music_kudu_online.music_kudu_internal.ads_ab_rtrs_user_metric_hour
SELECT
from_unixtime(`timestamp`, 'yyyy-MM-dd') as dt,
from_unixtime(`timestamp`, 'HH') as `hour`,
os, sceneid, parent_exp, `exp`, exp_type, userid,
count(1) pv
FROM iplay_ods.ods_rtrs_ab_log
INNER JOIN abtest_online.abtest.abtest_sence_metric_relation
FOR SYSTEM_TIME AS OF user_metric.proctime
ON ods_rtrs_ab_log.sceneid = abtest_sence_metric_relation.sceneid
GROUP BY from_unixtime(`timestamp`, 'yyyy-MM-dd'),
from_unixtime(`timestamp`, ‘HH’),
os, sceneid, parent_exp, `exp`, exp_type, userid 這是一個實時全聚合任務，在原始的FLINK中這段SQL執行的DAG大概是這樣的：
假如我們讀取的流表ods_rtrs_ab_log有5個分區，我們的SQL任務有七個并發，因為受到Kafka分區數的影響，加上FLINK本身作業鍊的優化，我們的消息的讀取、維表JOIN、MINI BATCH的操作全部受到了Kafka分區的影響，無法擴充，特别是對于維表JOIN這種IO操作來說，任務的并發度嚴重影響了整體程式的性能，這個時候我隻能通過擴容Kafka的分區數來提升性能。
但是這種操作非常重，而且很有可能會影響其它讀取這張流表的任務；為了解決這個問題，我們對Kafka的Connector做了一些改造，支援通過配置多添加一步Shuffle操作，比如在上面的配置當中我們添加了配置：
'connector.rebalance.keys' = 'sceneid,parent_exp,userid' 消息會在讀取以後按照sceneid,parent_exp,userid等字段進行hash分片，這樣大大提高了整體程式的性能擴充性，而且通過指定字段的keyBy操作，可以大大提高維表JOIN緩存的命中率，提高MINI BATCH的性能和效率。
除了以上配置以外，我們還支援添加随機的Rebalance操作、Rescale操作以及解析行為的拆解，來進一步提升整體程式性能的擴充，這裡需要注意的是額外Shuffle操作，會帶來更多線程和網絡開銷，在配置這些操作的同時需要同時關注機器的負載情況，添加額外的Shuffle操作雖然能提升程式的擴充性，但是由于額外網絡和線程開銷，如果機器本身性能不行的話，很有可能會适得其反，在相同的資源情況下性能變得更差，這點需要根據自己程式以及環境情況進行配置。
3. Kafka使用優化
随着流量的飛速增長Kafka的穩定性也是我們面臨的主要難題，包括Kafka的機櫃帶寬問題、跨機房帶寬問題、Kafka擴縮容的抖動問題、還有Kafka本身配置問題等等，基本上大家能遇到的問題我們都遇到了，為了解決以上問題我們做了以下工作：
3.1 開發鏡像服務，解決帶寬問題，保障高優先級任務
我們通過FLINK自己開發了一套鏡像服務，在不同的機房子產品間分别部署了一套Kafka叢集，通過鏡像服務同步兩套Kafak叢集的資料，主Kafka提供給比較重要P0級别的實時任務，其它不是特别重要的任務讀取鏡像叢集的資料。
我們通過Yarn Label技術，通過不同隊列的選擇來控制任務所在的機房，來減少跨機房帶寬的消耗，為了友善使用者切換不同的Kafka叢集，我們在Flink流表側也做了一些改造，支援一張流表同時挂載多個Kafka叢集，隻要通過簡單的配置就可以随意切換Kafka叢集，經過一輪任務整理和切換，Kafka帶寬使用情況有了大大的改善：
3.2 Kafka監控完善
在日常的工作中，我們發現很多開發對Kafka本身并不太了解，運維由于經驗的不足在初期對整體Kafka的管控也不是那麼的嚴格，導緻在使用上有很多問題。是以我們整合了音樂内部的Kafka監控服務的資料，結合我們平台的任務血緣，開發了自己的一套Kafka監控服務。
目前這套系統整體還比較初級，除了關聯了Kafka、流表、和任務之間的關系以外，我們還對以下這幾種情況做了主動監控：
- Kafka Topic的分區數的合理性，主要監控消息隊列分區數過少或者過多的情況，主要是過少的情況，防止因為分區數過小，下遊任務處理性能跟不上的問題；
- Kafka分區資料生産均衡問題：防止因為Kafka本身分區資料的不均衡導緻下遊任務處理性能不行的問題；
- Kafka分區資料消費均衡問題：防止因為Kafka本身分區發生變化，而下遊任務因為沒有開啟分區感覺，導緻一些資料沒有消費到等問題；
- 流量激增和激降報警：關鍵隊列流量報警，保障實時資料的品質。
Kafka版本更新：為了解決本身Kafka擴容的穩定性問題、資源隔離問題，通過我們音樂公共技術團隊，在Kafka 2.X版本基礎上做了一些二次開發工作，将Kafka整個服務做了平台化的支援，支援了Topic的平滑擴所容，支援資源隔離。
類似YARN的LAEBL技術，支援針對不同的TOPIC劃分不同region的機器，完善的消息鏡像服務，且支援offset的複制；統一的Kafka運維監控平台，此部分内容後續文章會詳細介紹。
3.3 分區流表技術建設
實時數倉上線以後，我們發現以下幾種情況非常影響程式的穩定性以及流表的易用性：
- 很多時候我們隻需要一張流表中1%的資料，但是因為沒有辦法按需讀取，是以我們必須消耗大量的資源去解析讀取另外99%的資料，導緻了大量的資源帶寬的消耗，浪費了大量的資源，而且本身SQL的開發方式本身沒有辦法按需解析日志，導緻我們必須完整的解析出每一條消息，這就導緻進一步的計算資源的消耗。
- 當我們按照經驗和業務，将大的TOPIC拆分成很多小的TOPIC時，一張表變成了很多小表，使用者又必須有很多的經驗知識去了解這些schema完全相同的小表中分别包含了哪些消息，易用性很差，這樣的設計也不符合數倉的整體設計邏輯，以後如果要做批流表統一進制資料的時候，整體也變得不太可能
在離線場景下我們很有很多手段來解決以上問題，減少不必要的IO，如資料的分桶、存儲有序的資料利用Parquet的下推查詢的能力、做分區表等手段都可以解決以上問題。但是實時表的Case下在現有的公開的方案中好像并沒有什麼好的方法；是以為了解決以上問題，我們開發了流表的分區方案，整體和HIVE表的分區實作思想差不多：
我們使用Flink Table Souce提供的SupportsFilterPushDown的接口實作了一套自己的實時流表分區方案，一個分區對應一個topic，通過使用者的查詢條件下推過濾掉沒有必要的分區，進而減少沒有必要的資料的讀取；目前已經上線了第一版，初步拆分了雲音樂曝光日志，順便還嘗試使用AVRO的資料格式代替以前的JSON格式，實踐下來優化效果明顯：
- 使用AVRO格式格式基本都能帶來至少30+%的的帶寬優化，消息解析性能相對音樂的原始日志格式的解析性能提升一倍.
- 使用分區流表，我們初步遷移了了4個曝光日志的消費任務，已經節省了7台實體機，平均節省計算和帶寬資源75%以上。
雖然這些都是比較極端的Case，但是從這些例子我們可以預計分區流表技術全面鋪開以後，使用得到的話，絕對是一個能帶來質變的優化。
二、批流一體
資料實時化一直是我們雲音樂資料平台團隊數倉建設的一個比較大的目标，在這個目标的背後批流一體也是我們繞不開一個“名詞”、“概念”、“技術”、或者是個“産品”。在正式開始分享我們的工作以前，首先分享下我有一次在電梯間遇到算法同學，然後和算法同學發生的對話：
算法：你們的批流一體什麼時候上線？我們等着用呢？
我： 你們目前的訴求是什麼呢？
算法：我們現在很多實時名額都是自己開發，沒法在離線以後直接使用現成數倉資料。
從這段對話我們可以看出，算法同學并不是想要什麼批流一體的技術，他們想要的是實時的現成的可用的數倉資料，來提升他們的開發效率，批流一體的背後，不同角色的業務方的訴求是什麼呢？
對于營運、産品、老闆、分析師們來說：
他們想要看到的是準确的實時的可分析的報表資料，關鍵點在于可分析上。當結果資料發生異常波動時，我們得有實時的明細資料提供分析查詢，來調查發生異常波動的原因。當老闆有一些新的想法，想對現成的報表做下二次分析時，我們得有能力提供明細的可分析的資料來做分析給出結果。
以實時日活統計來說，我們常用的手段是将使用者ID存儲的Redis這樣KV存儲當中來做去重，或者近似去重，然後計算得出實時的日活資料，但是當日活發生異常波動時，因為Reids的資料不是可分析的。是以我們很難快速給出原因，也沒法在當天做分析，這種方案和結果顯然是不合格的。
對于數倉開發來說：
- 統一實時/離線數倉中繼資料管理、統一模型、統一存儲，減少數倉運維建設成本，提升整體數倉的易用性；
- 統一開發代碼，統一一套SQL解決離線/實時開發問題，降低開發運維成本，徹底解決因為業務了解不同、邏輯不同導緻的實時離線資料結果差異大的問題。
對于算法同學來說：
有實時/離線統一的數倉表可以可以用使用，統一模型，降低業務了解的門檻，提升整體數倉資料的易用性，友善好用的數倉中繼資料管理服務，友善算法同學進行二次的特征開發工作，提升模型的開發效率。提供準确實時可分析的算法模型效果資料，提升算法同學模型疊代的效率
整體總結下來批流一體的目标主要包含三個方面：
- 統一代碼：一套SQL完成實時和離線的相關業務的開發需求；
- 統一數倉中繼資料：一張表可以同時提供離線讀和實時讀，統一模型的批流一體的數倉；
- 實時的報表資料：這與統一數倉中繼資料不同，産品報表資料需要提供秒級的實時的結果的查詢能力，而統一數倉資料往往隻需要實時的存儲即可，對OLAP查詢的效率，并沒有報表資料并沒有那麼敏感。
1. 統一代碼
由于實時SQL本身并沒有特别的成熟，很多在離線場景下很容易實作的邏輯，在實時場景下要麼是不能實作，要麼是穩定性有問題。
目前業界都還在探索當中，阿裡目前主要的方式的是使用FLINK一套引擎解決實時離線統一SQL的問題，但是目前也都是在實踐，在上層ADS層業務邏輯實作上通過底層數倉的建設屏蔽掉一些實時SQL能力的問題，做到産品報表開發上統一一套SQL。這也是我們未來可以嘗試的方向，除了在上層報表開發上嘗試統一SQL以外，我們在統一代碼這一塊也做了一些工作和規劃：
- 統一UDF，內建更新平台架構到FLINK1.12新版本，統一離線實時統一套UDF；
- 統一進制資料管理：在FlinkSQL側我們繼承中繼資料中心服務，提供catalog.db.table這樣的資料讀取和寫入方式，為了統一進制資料，同樣我們對SparkSQL做了二次的封裝，同樣和中繼資料中心做了內建，實作了以catalog.db.table這樣形式的異構資料源之間的讀取和寫入。
場景化的配置式的批流一體的統一實作，對于一些簡單業務邏輯的場景，我們後續會開發場景化的批流一體的實作。如批流一體的索引任務、批流一體的ETL清洗平台等等，這塊由于資源問題，目前還在規劃中。
批流一體SQL統一的在目前的技術下，還有一個比較大的前提是本身日志的複雜程度，這個涉及到本身日志埋點規範性和完整性，實時計算不像離線，可以将大量歸因邏輯， 關聯邏輯放在資料側進行處理，抛開合理性和成本問題，很多工作在離線場景下是可以做的。
但是在實時場景，本身對性能和穩定性都非常的敏感，如果将大量的邏輯都放在資料側進行處理，本身就會帶來很多不能實作的問題、實作起來成本高的問題、很多穩定性、以及資料延遲的問題。如果打點做不好，整個實時數倉建設都是問題，是以雲音樂也啟動了曙光打點項目和有數團隊合作，徹底重構雲音樂各個産品的打點的實作，提升和完善打點的規範性和準确性，降低實時數倉的開發成本問題。
2. 統一數倉中繼資料
目前業界主要有兩類方案：
- 第一種是建設批流映射層的方案，目前阿裡公開的方案的就是這種方案，比較适合已經有了實時數倉和離線數倉的老産品，在不改動原有數倉的情況下，建構統一映射層視圖，通過視圖的方式提供一體化的使用體驗，整體的原理參考下圖：
2021 年網易雲音樂實時計算平台發展和挑戰
- 第二種方案是建構一種新的中繼資料系統，一套schema下同時挂載多種存儲，如HDFS、Kafka等，在寫入資料時同時寫入，在讀取場景下時，根據讀取方式的不同，選擇相應的合适的存儲，目前網易數帆有數産品團隊開發的Arctic采用的就是這種方案：
2021 年網易雲音樂實時計算平台發展和挑戰
整體思路是封裝icberg和Kafka以及Hbase等多種存儲，在不同場景下使用不同的存儲，另外arctic還在iceberg的基礎上做了很多二次開發，來解決DWS資料的更新問題，提供類似Hudi的CopyOnWrite以及MergeOnRead等功能，用來解決Flink本身用來做全聚合的穩定性問題。目前雲音樂已經在一些新的業務場景做了試用，已經上線了幾十張的的批流一體表，大家如果想進一步了解arctic可以找網易數帆有數實時計算團隊了解，在此不過多描述。
3. 實時的報表資料
提供實時的報表資料主要依賴OLAP引擎和存儲，存儲側需要有需要有在提供實時的資料更新能力的同時，還需要有提供秒級别資料的查詢能力，很多時候沒有辦法把将結果直接寫到到存儲中。因為資料報表本身很多靈活性的查詢，如果直接将結果寫到存儲中， 就需要類似Kylin那種實時的Cube能力，這對開發以及Flink本身計算的壓力太大， 本身也會帶來很多資源的和存儲的浪費，穩定性問題以及開發工作量的問題也會很多，資料的二次分析能力也會很局限；是以在這一層我們需要OLAP引擎提供至少百億級别的資料的秒級延遲的查詢的能力，目前我們主要的方案采用的存儲有Kudu和Clickhouse兩種，以我們老版本的ABTest為例，我們采用的方案如下：
對于實時的最新的小時次元以及天次元的結果我們通過Impala及時讀取Kudu資料關聯出最新的結果；對于曆史的一天以前天次元資料或者兩個小時以前小時次元的資料我們采用Spark預計算好存儲在結果表當中，兩份資料UNION在一起提供給使用者，保障資料結果的時效性，以及整體資料查詢的使用者體驗。
三、未來規劃
**運維工具的完善
**
實時SQL的發展降低了實時資料統計的開發難度，大大降低了實時資料統計的門檻，一方面由于本身實時SQL的不成熟而且黑盒，另一方面很多同學帶着離線SQL的開發經驗或者MYSQL類資料庫的SQL經驗來開發實時任務，這給平台帶來了很大的運維壓力，是以運維工具相關的建設，任務實時名額的完善是我們未來主要思考的方向之一。
分區流表技術完善
分區流表技術是一個能給雲音樂實時平台資源使用，Kafka壓力以及數倉建設帶來質變的技術，目前我們隻是完成了一個初版，未來我們會在分區的動态感覺，分區的修改， schema的修改，以及運維監控以及推廣上繼續完善。
場景化批流一體建設
如批流一體索引任務建設、批流一體ETL工具等, 統一日志清洗規則, 為批流一體數倉打好基礎。
批流一體存儲探索
- 調研業界目前的方案, 結合音樂的業務場景, 提供整套解決方案, 降低實時報表的開發門檻, 提升實時報表的開發效率；
- 批流一體邏輯層建設等。
最後附一張網易數帆有數團隊的實時計算解決方案架構圖，基于 Apache Flink 建構的高性能、一站式實時大資料處理方案，廣泛适用于流式資料處理場景。
Flink Forward Asia 2021
報名現已開放
Flink Forward Asia 2021 重磅啟動！FFA 2021 将于 12 月 4-5 日在北京·國家會議中心舉辦，預計将有 3000+ 開發者參與，探讨交流 Flink 最新動态。報名通道已開啟：
https://flink-forward.org.cn 更多 Flink 相關技術問題，可掃碼加入社群釘釘交流群
第一時間擷取最新技術文章和社群動态，請關注公衆号～
活動推薦
阿裡雲基于 Apache Flink 建構的企業級産品-實時計算Flink版現開啟活動：
99 元試用
實時計算Flink版（包年包月、10CU）即有機會獲得 Flink 獨家定制T恤；另包 3 個月及以上還有 85 折優惠！
了解活動詳情：
https://www.aliyun.com/product/bigdata/sc