一、愛奇藝大數(shù)據(jù)介紹

1.愛奇藝大數(shù)據(jù)應(yīng)用

圖片

愛奇藝大數(shù)據(jù)應(yīng)用分為兩個(gè)方面：

• 看過去：針對(duì)以往數(shù)據(jù)，制作天級(jí)別、小時(shí)級(jí)別或?qū)崟r(shí)的報(bào)表；負(fù)責(zé)內(nèi)容、會(huì)員的運(yùn)營工作，基于數(shù)據(jù)分析進(jìn)行決策。
• 知未來：基于數(shù)據(jù)制作用戶產(chǎn)品，應(yīng)用于實(shí)施廣告、推薦搜索等方面。

2.愛奇藝大數(shù)據(jù)實(shí)時(shí)應(yīng)用

愛奇藝大數(shù)據(jù)實(shí)時(shí)應(yīng)用包括實(shí)時(shí)廣告系統(tǒng)、實(shí)時(shí)推薦和搜索、實(shí)時(shí)熱度，以及實(shí)時(shí)風(fēng)控。下方右圖是應(yīng)用搜索界面，搜索推薦基于用戶及其他個(gè)性化信息實(shí)時(shí)計(jì)算生成。

圖片

3.愛奇藝大數(shù)據(jù)服務(wù)體系

圖片

以上提及的產(chǎn)品通過愛奇藝大數(shù)據(jù)服務(wù)體系實(shí)現(xiàn)。如上圖，愛奇藝大數(shù)據(jù)服務(wù)體系主要由數(shù)據(jù)采集、數(shù)據(jù)處理、數(shù)據(jù)應(yīng)用等方面構(gòu)成。

• 數(shù)據(jù)采集包括用戶和內(nèi)容數(shù)據(jù)、服務(wù)日志、監(jiān)控?cái)?shù)據(jù)、第三方數(shù)據(jù)；
• 數(shù)據(jù)處理包括大數(shù)據(jù)基礎(chǔ)架構(gòu)的存儲(chǔ)和計(jì)算層，以及上層的數(shù)據(jù)分析引擎和平臺(tái)；
• 最終通過報(bào)表、分析型查詢、機(jī)器學(xué)習(xí)和監(jiān)控排障功能，應(yīng)用于各個(gè)業(yè)務(wù)線。

我們的集群部署模式由通用集群和少量專用集群組成，總共有1萬多臺(tái)機(jī)器、500PB以上的存儲(chǔ)，每天會(huì)運(yùn)行50多萬個(gè)批處理任務(wù)及4000多個(gè)流計(jì)算任務(wù)。

二、愛奇藝實(shí)時(shí)計(jì)算技術(shù)演進(jìn)

1.愛奇藝實(shí)時(shí)計(jì)算技術(shù)演進(jìn)概覽

圖片

技術(shù)演進(jìn)過程分為三個(gè)階段：

第一階段是比較原始的Hive on MapReduce。在此階段，我們借助Hive工具實(shí)現(xiàn)了SQL化的分析，通過SQL在Hadoop上構(gòu)建離線數(shù)倉。SQL避免用戶自己寫MapReduce的Java代碼，解決了大數(shù)據(jù)的初步問題之一。但隨著業(yè)務(wù)發(fā)展、實(shí)時(shí)性需求增加，離線分析的處理延時(shí)難以滿足業(yè)務(wù)需要。

第二階段采用基于Spark SQL分析，大大加速離線數(shù)倉的構(gòu)建進(jìn)程，同時(shí)也探索基于Flink的實(shí)時(shí)數(shù)據(jù)處理。這個(gè)階段我們初步引入Flink，用戶直接寫Flink任務(wù)代碼，相比于基于SQL的離線分析，開發(fā)和運(yùn)維的難度高了不少。同時(shí)，因?yàn)樾枰S護(hù)離線和實(shí)時(shí)兩條鏈路，成本較高，且存在流與批數(shù)據(jù)不一致等問題。

第三階段主要進(jìn)行兩項(xiàng)工作：一方面是實(shí)時(shí)計(jì)算SQL化，我們引入了統(tǒng)一元數(shù)據(jù)，簡(jiǎn)化了Flink SQL的開發(fā)，使得撰寫實(shí)時(shí)計(jì)算的邏輯與Hive SQL一樣簡(jiǎn)單；另一方面，引入數(shù)據(jù)湖Iceberg，初步實(shí)現(xiàn)流批一體。

2.Spark SQL服務(wù)架構(gòu)

愛奇藝的Spark SQL服務(wù)基于開源的Apache Kyuubi搭建，因?yàn)橹苯邮褂肧park Thrift Server服務(wù)有很多缺點(diǎn)，比如不支持多租戶、資源隔離較難實(shí)現(xiàn)等。

圖片

引入Kyuubi后，整體架構(gòu)如上圖所示。我們?cè)贖adoop集群上層搭建了Kyuubi Server集群，再上層通過Pilot統(tǒng)一SQL網(wǎng)關(guān)（自研服務(wù)）接入，最上層是離線計(jì)算的Gear工作流調(diào)度系統(tǒng)和魔鏡即席查詢平臺(tái)，分別承接定時(shí)工作流以及Ad-hoc的即席查詢。

除此之外，Kyuubi Server和Spark任務(wù)引擎會(huì)注冊(cè)到ZooKeeper服務(wù)發(fā)現(xiàn)集群中，供其調(diào)用方進(jìn)行服務(wù)發(fā)現(xiàn)，由此實(shí)現(xiàn)了高可用性，去除了單點(diǎn)故障。

基于Kyuubi的這套體系具有以下6個(gè)好處：

• 多租戶：企業(yè)內(nèi)部包含各個(gè)業(yè)務(wù)線和不同的Hadoop user，只需部署一套服務(wù)即可支持不同用戶的訪問；
• 資源隔離：我們?yōu)槊總€(gè)來自定時(shí)工作流的例行化任務(wù)，啟動(dòng)獨(dú)立的Spark引擎，各個(gè)任務(wù)間資源隔離，不會(huì)互相影響；
• 快速Ad-hoc支持：我們也為來自即席查詢的任務(wù)，啟動(dòng)用戶級(jí)別的共享Spark引擎，無需在每次查詢時(shí)都花費(fèi)1分鐘去啟動(dòng)Spark，使得查詢響應(yīng)低至秒級(jí)；
• 高擴(kuò)展性：?jiǎn)蝹€(gè)Kyuubi實(shí)例支持上千個(gè)連接，且每個(gè)查詢只需要花費(fèi)Kyuubi實(shí)例很少的資源，因此每臺(tái)Kyuubi服務(wù)器都可承擔(dān)很高的工作負(fù)載；
• 高可用性：由于采用了ZooKeeper的服務(wù)注冊(cè)與發(fā)現(xiàn)機(jī)制，單個(gè)Kyuubi實(shí)例故障不影響任務(wù)運(yùn)行；
• 執(zhí)行計(jì)劃優(yōu)化：為了加快SQL的運(yùn)行速度，Kyuubi默認(rèn)支持了一些查詢優(yōu)化規(guī)則，并且允許用戶在此基礎(chǔ)上擴(kuò)展，添加新的優(yōu)化規(guī)則。

愛奇藝在作為Apache Kyuubi用戶的同時(shí)，也積極參與社區(qū)討論，回饋社區(qū)，目前共有70多個(gè)patches被社區(qū)接受。

3.Spark深入優(yōu)化

圖片

除了通過Kyuubi建立Spark SQL的服務(wù)之外，我們也對(duì)Spark本身進(jìn)行了優(yōu)化，使得計(jì)算速度更快，資源更節(jié)省。

• 動(dòng)態(tài)資源分配（Dynamic Resource Allocation，DRA）

由于我們平臺(tái)上每天都會(huì)運(yùn)行大量的任務(wù)，用戶很難為每個(gè)任務(wù)配置一個(gè)合適的資源量，因此經(jīng)常出現(xiàn)任務(wù)的并行度不足或資源浪費(fèi)的問題。DRA是Spark提供的已經(jīng)比較成熟的功能，開啟之后能夠根據(jù)并行度需求自動(dòng)申請(qǐng)或釋放資源，在避免資源浪費(fèi)的同時(shí)，還能在一定程度上加快任務(wù)運(yùn)行。

• 自適應(yīng)查詢執(zhí)行（Adaptive Query Execution，AQE）

AQE的定義是根據(jù)Spark任務(wù)運(yùn)行時(shí)的數(shù)據(jù)，動(dòng)態(tài)修改查詢計(jì)劃。因此它是一個(gè)優(yōu)化框架，而非特定功能，用戶可以擴(kuò)展各種優(yōu)化規(guī)則。

社區(qū)版Spark自帶的優(yōu)化規(guī)則包括Shuffle分區(qū)合并、自動(dòng)轉(zhuǎn)換為廣播Join、Join傾斜優(yōu)化等。我們基于Kyuubi進(jìn)行功能擴(kuò)展，比如自動(dòng)合并小文件、末級(jí)Stage配置隔離等。其中，末級(jí)Stage配置隔離是一個(gè)非常好用的優(yōu)化規(guī)則，它允許在配置層面，為普通Stage和末級(jí)Stage分別配置處理并行度。

這樣，我們可以在末級(jí)Stage上按目標(biāo)文件大小設(shè)置并行度，以合并小文件；在普通Stages上配置較高的并行度，加速任務(wù)處理，達(dá)成兩者兼顧的效果。

• 血緣集成Apache Atlas

愛奇藝內(nèi)部使用Apache Atlas管理數(shù)據(jù)血緣，為此我們將其元數(shù)據(jù)和血緣投遞的邏輯集成到了Kyuubi中，使得Kyuubi在運(yùn)行Spark SQL任務(wù)時(shí)，能夠自動(dòng)向Atlas投遞血緣。我們已將這一功能貢獻(xiàn)給了社區(qū)（KYUUBI #4814），將在即將發(fā)布的Kyuubi V1.8版本中可用。

• Apache Uniffle：Remote Shuffle Service

在Spark中使用Remote Shuffle Service是近兩年來比較流行的一個(gè)趨勢(shì)。愛奇藝采用的是Apache Uniffle這個(gè)開源產(chǎn)品。

在引入Apache Uniffle前，存在兩種問題：一個(gè)是Shuffle不穩(wěn)定，比如大數(shù)據(jù)量情況下，下載數(shù)據(jù)失敗，出現(xiàn)fetch failure的報(bào)錯(cuò)；另一個(gè)是存算分離的云原生架構(gòu)下，計(jì)算節(jié)點(diǎn)容量、IO性能不足。

引入Apache Uniffle后，原有問題得到改善：

• 減小磁盤IO壓力：利用內(nèi)存存儲(chǔ)數(shù)據(jù)，避免隨機(jī)IO
• 增強(qiáng)大數(shù)據(jù)量下的穩(wěn)定性：Shuffle 10TB以上不再失敗，可平穩(wěn)運(yùn)行
• 提高Shuffle速度：1TB TeraSort 快30%+

愛奇藝作為Apache Uniffle的共同貢獻(xiàn)者，深度參與了社區(qū)討論和貢獻(xiàn)。歡迎大家試用并提出反饋意見。

4.從Hive遷移Spark SQL

圖片

在支持Spark SQL后，已有的大量Hive任務(wù)需要遷移過來。遷移過程會(huì)遇到兩種問題：

• 運(yùn)行失?。篠park SQL的語法并不是100%兼容Hive，其語法更加嚴(yán)格；
• 數(shù)據(jù)不一致：相比于運(yùn)行失敗更加麻煩，因?yàn)橛脩魺o法立即發(fā)現(xiàn)，且對(duì)比數(shù)據(jù)一致性的工作也更加復(fù)雜。

為了解決遷移的問題，我們基于Pilot SQL網(wǎng)關(guān)開發(fā)了“雙跑對(duì)數(shù)”的功能，在遷移前自動(dòng)預(yù)測(cè)遷移結(jié)果，運(yùn)行步驟如下：

• SQL解析：輸入、輸出 
• 創(chuàng)建影子表：用戶模擬雙跑的輸出 
• 模擬運(yùn)行：修改SQL的輸出表，指定Hive、Spark引擎分別運(yùn)行 
• 結(jié)果一致性校驗(yàn)：行數(shù)、CRC32（浮點(diǎn)型保留小數(shù)點(diǎn)后4位，避免因精度不同導(dǎo)致的判斷錯(cuò)誤）

使用“雙跑對(duì)數(shù)”功能之后，我們?cè)谶w移的過程中發(fā)現(xiàn)了一些問題，其中有部分可以通過優(yōu)化Spark SQL的兼容性來解決，進(jìn)一步降低用戶遷移的工作量：

• 支持UDAF/UDTF的私有化構(gòu)造 
• 允許reflect拋出異常時(shí)返回NULL
• Hive參數(shù)映射到Spark參數(shù)，如 mapreduce.job.reduces→Spark.sql.shuffle.partitions

用戶在Hive中設(shè)置很多參數(shù)，比如reduce的個(gè)數(shù)，但這些參數(shù)在Spark中原本無法被識(shí)別，我們通過參數(shù)映射，將其轉(zhuǎn)化為Spark的相應(yīng)參數(shù)，盡可能保留用戶的SQL邏輯。

最后，使用“自動(dòng)降級(jí)”功能令遷移順利進(jìn)行，即首次使用Spark運(yùn)行失敗后，重試時(shí)降級(jí)為Hive引擎。由此，遷移分為兩個(gè)階段：第一階段開啟自動(dòng)降級(jí)，用戶可以放心遷移，并通過降級(jí)的記錄梳理出遷移失敗的任務(wù)；第二階段，將這些失敗的任務(wù)修復(fù)后，再完全切換到Spark。

目前Hive遷移的總體進(jìn)度已經(jīng)達(dá)成90%，對(duì)于這些遷移的任務(wù)，平均性能提升了67%，資源（包括CPU、內(nèi)存使用量）也降低了近一半。

5.Flink SQL + 統(tǒng)一元數(shù)據(jù)中心

圖片

在使用Spark SQL提高實(shí)時(shí)性的同時(shí)，我們也嘗試引入Flink SQL，希望能夠真正做到實(shí)時(shí)計(jì)算。但原生的Flink SQL如上示左圖，比Hive SQL長(zhǎng)很多，需要定義輸入輸出表，字段名稱和類型，以及背后的數(shù)據(jù)源配置。應(yīng)如何解決使用過程繁瑣的問題？

我們引入了“統(tǒng)一元數(shù)據(jù)中心”的概念，類似于Hive的Metastore。因?yàn)镠ive具有Metastore，所以無需反復(fù)定義輸入輸出表，寫SQL非常簡(jiǎn)單，如上圖中寫三行語句即可。

我們將內(nèi)部的各種數(shù)據(jù)，包括流式的Kafka和RocketMQ，傳統(tǒng)數(shù)據(jù)庫MySQL、Redis、HBase，以及數(shù)據(jù)湖產(chǎn)品，都集成到統(tǒng)一元數(shù)據(jù)中心，并開發(fā)了Flink Catalog、Flink Connectors與其對(duì)接。這樣依賴，我們無需在每個(gè)任務(wù)中，重新定義表的結(jié)構(gòu)以及連接串等信息，做到開箱即用，有效提升開發(fā)效率。

6.SQL適合流計(jì)算開發(fā)

可能有同學(xué)會(huì)有疑問，SQL到底能否足夠表達(dá)流計(jì)算？

因?yàn)閭鹘y(tǒng)SQL（比如Hive、MySQL），輸入是一個(gè)表，輸出也是一個(gè)表，從表到表的SQL究竟能否表達(dá)流式的計(jì)算邏輯？我們認(rèn)為是可以的。

這個(gè)觀點(diǎn)具有理論支撐，一位來自Google的工程師Akidau在其著作《Streaming Systems》中，提出了流和表的“相對(duì)論”。他認(rèn)為流和表本質(zhì)上是數(shù)據(jù)的兩種表現(xiàn)形式。他拆解了傳統(tǒng)SQL表到表的過程，將其拆分為表到流、流到表、流到流三種操作的組合。

以上圖右邊的SQL舉例，輸入是一組用戶得分，按照?qǐng)F(tuán)隊(duì)進(jìn)行聚合，計(jì)算出每個(gè)團(tuán)隊(duì)的總分，輸出到新的表中。它的輸入表由4個(gè)字段組成：用戶、團(tuán)隊(duì)、得分和時(shí)間。

讓我們來拆解這個(gè)SQL的執(zhí)行邏輯（假設(shè)這是離線計(jì)算）。首先，原始表并不是一次性加載到內(nèi)存的，而是通過一個(gè)SCAN算子，一條一條地讀入，變成內(nèi)部的流。然后經(jīng)過SELECT算子，去掉無用字段，保留team和score字段，得到了一個(gè)新的流。

最后，流的數(shù)據(jù)全部到齊后，一次性計(jì)算聚合的值，即把每個(gè)team的所有分?jǐn)?shù)相加得到總分，再輸出到目標(biāo)表。由此看出，第一個(gè)操作SCAN是表到流，第二個(gè)操作SELECT是流到流，第三個(gè)操作GROUP BY是流到表。

從上面的SQL執(zhí)行邏輯拆解可以看出，將傳統(tǒng)SQL描述為表到表的操作，黑盒地看是對(duì)的，但在微觀層面是不準(zhǔn)確的，實(shí)際上是表到流、流到表、流到流三種操作的組合，唯一不存在的是直接的流到流的操作。

流計(jì)算的過程中包括很多要素，比如Map或Filter可以認(rèn)為是一個(gè)流到流的操作，分組的聚合或窗口的聚合，就是流到表的過程；而通過定時(shí)的trigger或Watermark引起的trigger，是表到流的過程。

當(dāng)把上面的SQL看成流計(jì)算時(shí)，會(huì)發(fā)現(xiàn)其拆解過程與離線計(jì)算一模一樣，都是由SCAN（表到流）、SELECT（流到流）、GROUP BY（流到表）組成的。

因此，SQL對(duì)于流計(jì)算和離線計(jì)算來說，沒有本質(zhì)區(qū)別，所以它非常適合流計(jì)算的開發(fā)。SQL開發(fā)優(yōu)勢(shì)如下：

• 開發(fā)門檻低：相較于通過Java/Scala代碼開發(fā)Flink任務(wù)，SQL的開發(fā)門檻較低。引入開箱即用的元數(shù)據(jù)后，SQL就更加簡(jiǎn)潔，用戶無需學(xué)習(xí)Flink的API。
• 版本升級(jí)容易：Flink SQL對(duì)齊了SQL標(biāo)準(zhǔn)，語法相對(duì)穩(wěn)定，跨版本升級(jí)改動(dòng)較小。
• 運(yùn)行效率高：因?yàn)镕link SQL具有一些參數(shù)，控制SQL執(zhí)行的計(jì)劃優(yōu)化，無需復(fù)雜的代碼邏輯實(shí)現(xiàn)這些功能。

在愛奇藝的實(shí)時(shí)計(jì)算平臺(tái)上，目前SQL的任務(wù)占比已經(jīng)達(dá)到2/3，已經(jīng)能覆蓋大部分的功能，所以較推薦內(nèi)部用戶使用SQL進(jìn)行流計(jì)算的開發(fā)。

7.Lambda到流批一體架構(gòu)

圖片

我們?cè)诖鎯?chǔ)側(cè)也做了技術(shù)革新。傳統(tǒng)方案使用Lambda架構(gòu)，即離線一條通路、實(shí)時(shí)一條通路，在下游合并這兩條通路。但這種架構(gòu)存在明顯問題：

• 離線通路時(shí)效性差，實(shí)時(shí)通路容量低
• 維護(hù)兩套邏輯，開發(fā)效率低
• 兩條通路數(shù)據(jù)不一致
• 維護(hù)多套服務(wù)，成本高

我們通過引入數(shù)據(jù)湖技術(shù)，可以做到流批一體架構(gòu)，即使用Flink與數(shù)據(jù)湖交互，實(shí)時(shí)寫入、實(shí)時(shí)更新。數(shù)據(jù)湖技術(shù)解決了兩條鏈路、實(shí)時(shí)性、以及實(shí)時(shí)通路容量不足的問題。由于無需維護(hù)兩條通路，計(jì)算成本與存儲(chǔ)成本比之前的模式更低。

8.基于Iceberg的數(shù)據(jù)湖

圖片

愛奇藝選擇的數(shù)據(jù)湖產(chǎn)品是Apache Iceberg，其具體好處將通過案例介紹。

上圖是會(huì)員訂單分析的應(yīng)用場(chǎng)景。愛奇藝的會(huì)員業(yè)務(wù)有10多年的歷史，每個(gè)會(huì)員訂單都對(duì)應(yīng)一條記錄，訂單表存儲(chǔ)在MySQL中，這些表非常大。會(huì)員團(tuán)隊(duì)進(jìn)行用戶會(huì)員運(yùn)營分析時(shí)，如果直接用MySQL對(duì)這些表進(jìn)行查詢，速度非常慢，因?yàn)镸ySQL對(duì)這種OLAP分析的場(chǎng)景支持不佳。

最原始的方案是通路1（上圖標(biāo)號(hào)1和2），先用內(nèi)部數(shù)據(jù)集成工具BabelX將MySQL表全量導(dǎo)出到Hive，再使用Hive、Spark SQL或Impala查詢。這條通路的問題是，MySQL的全量導(dǎo)出是一個(gè)天級(jí)別的任務(wù)，數(shù)據(jù)分析的時(shí)效性很差；每次導(dǎo)出的數(shù)據(jù)量很大，對(duì)MySQL產(chǎn)生很大壓力；每天都在反復(fù)導(dǎo)出相同的數(shù)據(jù)，效率很低。

后來會(huì)員團(tuán)隊(duì)和我們合作了另外一條通路2（即上圖標(biāo)號(hào)3和4），通過內(nèi)部工具，將MySQL的變更流實(shí)時(shí)導(dǎo)出到Kudu，用Impala進(jìn)行查詢。Kudu介于HDFS和HBase之間，既有實(shí)時(shí)寫入的能力，又有分析型查詢能力。這條通路的問題在于：

• Impala+Kudu需要搭建單獨(dú)的集群，成本比較高；
• Kudu集群的擴(kuò)展性差，因?yàn)榧荷暇€只能達(dá)到幾十個(gè)節(jié)點(diǎn)；
• Impala+Kudu運(yùn)維比較復(fù)雜，經(jīng)常出現(xiàn)故障。

基于這些痛點(diǎn)，我們調(diào)研后發(fā)現(xiàn)Iceberg比較適合完成這種任務(wù)，我們選擇了圖中最下面的新通路：通過內(nèi)部的RCP平臺(tái)，使用Flink CDC技術(shù)實(shí)時(shí)導(dǎo)出到Iceberg中，在下游使用Spark SQL進(jìn)行查詢。

改造效果如下：

• 時(shí)延低：整體時(shí)延大幅降低，從原來的天級(jí)延遲可以降低到分鐘級(jí)
• 查詢快：我們對(duì)查詢性能進(jìn)行了優(yōu)化，使其達(dá)到了接近Impala+Kudu的查詢性能
• 成本低：利用現(xiàn)成的HDFS集群，無需獨(dú)立部署
• 運(yùn)維易：對(duì)MySQL壓力較小，鏈路穩(wěn)定，運(yùn)維工作量也較小

在查詢性能上，我們做了兩處優(yōu)化：

1）小文件智能合并：Iceberg表在寫入過程中會(huì)產(chǎn)生很多小文件，積累到一定程度會(huì)嚴(yán)重影響查詢性能；而合并小文件時(shí)，如果每次都全表合并，又會(huì)造成嚴(yán)重的寫放大。為此我們開發(fā)了智能合并策略，基于分區(qū)下文件大小均方差，自動(dòng)選擇待合并的分區(qū)，最大程度地避免了寫放大。

2）寫Parquet文件開啟BloomFilter：BloomFilter可以判斷一組數(shù)據(jù)中是否不含指定數(shù)據(jù)，被Parquet等存儲(chǔ)格式廣泛使用，用來降低讀取數(shù)據(jù)量。愛奇藝將這一特性集成到Iceberg中，在寫Parquet文件時(shí)允許開啟BloomFilter，在內(nèi)部場(chǎng)景中取得了很好的效果。這一功能已貢獻(xiàn)給社區(qū)（PR #4831）。

最終，查詢的時(shí)間從900秒降低到10秒，達(dá)到了交互式查詢的性能，很好地滿足了會(huì)員運(yùn)營分析的需求。

三、愛奇藝實(shí)時(shí)計(jì)算平臺(tái)建設(shè)

1.平臺(tái)建設(shè)與數(shù)據(jù)處理架構(gòu)

圖片

愛奇藝的實(shí)時(shí)計(jì)算主要又兩個(gè)平臺(tái)：負(fù)責(zé)通用型計(jì)算任務(wù)的RCP實(shí)時(shí)計(jì)算平臺(tái)、負(fù)責(zé)特定分析型需求的RAP實(shí)時(shí)分析平臺(tái)。

基于原始數(shù)據(jù)，可通過RCP進(jìn)行通用分析，將結(jié)果寫入新的流、數(shù)據(jù)庫或Iceberg，供線上服務(wù)和數(shù)據(jù)分析直接使用。如需根據(jù)事件流，制作實(shí)時(shí)報(bào)表等特定的復(fù)雜目標(biāo)，可使用RAP平臺(tái)。

2.RCP實(shí)時(shí)計(jì)算平臺(tái)

圖片

RCP（Real-time Computing Platform，愛奇藝統(tǒng)一實(shí)時(shí)計(jì)算平臺(tái)）的特點(diǎn)是：

• 流程完整：具備數(shù)據(jù)讀入、計(jì)算及分發(fā)的全流程
• 支持多種開發(fā)模式：JAR/SQL/DAG，其中DAG模式是對(duì)SQL模式的進(jìn)一步簡(jiǎn)化，通過圖形化界面配置即可完成開發(fā)
• 集成統(tǒng)一元數(shù)據(jù)中心：各種類型的數(shù)據(jù)源均由平臺(tái)統(tǒng)一注冊(cè)和管理

如上圖架構(gòu)所示，Server層負(fù)責(zé)資源管理、任務(wù)管理、任務(wù)提交、監(jiān)控報(bào)警等功能。Launcher層負(fù)責(zé)直接提交任務(wù)到運(yùn)行集群，這一層包含內(nèi)部的Flink版本和Spark版本，對(duì)于Flink，又包含了JAR/SQL/DAG引擎、接入統(tǒng)一元數(shù)據(jù)、以及各種數(shù)據(jù)源的connector。

1）接入傳統(tǒng)數(shù)據(jù)庫

圖片

RCP平臺(tái)能結(jié)合各個(gè)數(shù)據(jù)庫的Connnector，將傳統(tǒng)數(shù)據(jù)庫接入實(shí)時(shí)計(jì)算。

上圖是針對(duì)廣告庫存計(jì)算的實(shí)時(shí)化改造。業(yè)務(wù)需要對(duì)多個(gè)MySQL表做Join，寫入Redis中，供下游的實(shí)時(shí)任務(wù)查詢。

原有方案是，使用Spark批處理作業(yè)，每10分鐘全量拉取這些MySQL表，在Spark任務(wù)里進(jìn)行Join。這個(gè)方案的問題是，每10分鐘進(jìn)行全量拉表，對(duì)MySQL壓力較大，且整體寫入Redis的數(shù)據(jù)時(shí)效性較差，至少延遲10分鐘，這會(huì)導(dǎo)致業(yè)務(wù)數(shù)據(jù)的準(zhǔn)確性下降。

改造后的方案見圖中綠框，我們采用Flink CDC的方案。在Flink任務(wù)中配置三個(gè)CDC的source，由此實(shí)現(xiàn)對(duì)MySQL全量同步以及自動(dòng)轉(zhuǎn)增量拉取的過程；緊接著一個(gè)Join節(jié)點(diǎn)，負(fù)責(zé)實(shí)時(shí)計(jì)算Join結(jié)果。如此一來，Join的輸出是實(shí)時(shí)更新的結(jié)果，上游MySQL表的更新會(huì)實(shí)時(shí)地反映到Redis中。

改造效果：

• 將以上三個(gè)實(shí)時(shí)的表Join數(shù)據(jù)寫入Redis后，業(yè)務(wù)準(zhǔn)確性明顯提升，達(dá)到7個(gè)9；
• 時(shí)效性方面，從20分鐘左右降低到秒級(jí)，降低了MySQL的服務(wù)壓力；
• 由于沒有重復(fù)的數(shù)據(jù)處理，資源節(jié)省了50%，處理效率大幅提升。

RCP支持了各類CDC connector，降低了將數(shù)據(jù)庫接入實(shí)時(shí)計(jì)算的門檻，主要的優(yōu)勢(shì)有：

• 可以實(shí)現(xiàn)存量+增量的無縫對(duì)接
• 對(duì)源庫影響較小
• 相較其他實(shí)時(shí)同步方案，F(xiàn)link方案可以實(shí)現(xiàn)讀取和數(shù)據(jù)加工的一體性，比如無需借助Kafka實(shí)時(shí)隊(duì)列，整個(gè)鏈路更加簡(jiǎn)單。

2）故障診斷

RCP平臺(tái)支持故障診斷功能。針對(duì)單個(gè)任務(wù)，平臺(tái)可自助排查故障原因，如下圖所示：

• 日志分析：報(bào)錯(cuò)信息，比如checkpoint失敗次數(shù)過多
• 指標(biāo)分析：GC、流量?jī)A斜指標(biāo)等

圖片

3）鏈路血緣

如下圖所示，平臺(tái)展示了該任務(wù)上游、下游的血緣關(guān)系。

圖片

4）鏈路診斷

如果需要分析整條鏈路，平臺(tái)也提供了一鍵鏈路診斷的功能。只需點(diǎn)擊一下，即可對(duì)鏈路上的所有實(shí)時(shí)計(jì)算作業(yè)，進(jìn)行健康度情況分析，獲取其最近的重啟次數(shù)和消費(fèi)延遲等信息。

圖片

3.RAP實(shí)時(shí)分析平臺(tái)

圖片

愛奇藝RAP（Real-time Analytics Platform）實(shí)時(shí)分析平臺(tái)，提供一站式的大數(shù)據(jù)攝取、計(jì)算和分析能力，支持超大規(guī)模實(shí)時(shí)數(shù)據(jù)多維度的分析，并生成分鐘級(jí)延時(shí)的可視化報(bào)表。主要特色是：

• 提供實(shí)時(shí)、多維度的聚合分析
• 配置化：通過頁面配置完成大部分功能，相較實(shí)時(shí)計(jì)算平臺(tái)更簡(jiǎn)單
• 實(shí)時(shí)報(bào)表：自動(dòng)生成的可視化報(bào)表（內(nèi)部BI平臺(tái)+Grafana）
• 實(shí)時(shí)報(bào)警、數(shù)據(jù)接口

RAP的架構(gòu)包含4個(gè)模塊：

• 數(shù)據(jù)源接入方面，支持公司內(nèi)部各流式數(shù)據(jù)源；
• 數(shù)據(jù)處理方面，通過RCP平臺(tái)或Druid KIS進(jìn)行；
• 兩個(gè)OLAP引擎包括Druid和ClickHouse；
• 最后將可視化報(bào)表發(fā)布在Grafana或內(nèi)部BI平臺(tái)。

1）典型案例

圖片

上圖是一張直播報(bào)表，展示直播實(shí)時(shí)的卡頓比（HCDN團(tuán)隊(duì)）情況。

右圖是直播期間每分鐘的總UV值（同步在線人數(shù)），只需三個(gè)步驟就能完成該報(bào)表的配置：

• 首先選擇接入數(shù)據(jù)源：此處選擇一張Iceberg表（用戶行為表），包含時(shí)間、設(shè)備ID、事件類型、app版本、運(yùn)營商、省份等字段。
• 然后配置模型：指明時(shí)間戳字段，計(jì)算哪些指標(biāo)，哪些字段是維度
• 最后報(bào)表配置：指明用戶想要的報(bào)表展現(xiàn)方式

總體來說，只需少量的頁面操作，即可配置一張實(shí)時(shí)報(bào)表，整個(gè)過程非常迅速。原先使用通用型工具開發(fā)此類報(bào)表，可能需要一周時(shí)間，但在RAP進(jìn)行配置，僅需小時(shí)級(jí)別的時(shí)間，并且支持靈活的需求變更。目前，RAP平臺(tái)已在愛奇藝的直播、會(huì)員監(jiān)控等業(yè)務(wù)中廣泛應(yīng)用。

四、未來展望

下一步，愛奇藝實(shí)時(shí)計(jì)算的發(fā)展方向包括：

• 實(shí)時(shí)計(jì)算SQL化：進(jìn)一步提升SQL化的比例，豐富 SQL化的配套支持，比如調(diào)試功能，增強(qiáng)使用SQL開發(fā)的便捷性。
• 實(shí)時(shí)化的數(shù)據(jù)集成平臺(tái)：基于配置化的方式，完成同種數(shù)據(jù)源的不同集群、甚至不同種類數(shù)據(jù)源之間的數(shù)據(jù)同步。
• 流批一體新方案探索：跟進(jìn)社區(qū)新動(dòng)向，比如Hybrid Source和Apache Paimon等流批一體的存儲(chǔ)和計(jì)算產(chǎn)品。

Q&A

Q1：實(shí)時(shí)計(jì)算平臺(tái)后續(xù)演進(jìn)規(guī)劃是怎樣的？

A1：進(jìn)一步提升SQL化開發(fā)的成熟度，優(yōu)化調(diào)試和診斷功能，對(duì)Flink SQL進(jìn)行性能優(yōu)化，流批一體。

Q2：數(shù)據(jù)服務(wù)支持實(shí)時(shí)計(jì)算的同時(shí)，能否保存所有數(shù)據(jù)？存儲(chǔ)有期限嗎？

A2：可以，Iceberg存儲(chǔ)利用HDFS集群實(shí)現(xiàn)，其容量非常大。但用戶仍需要配置期限，無論何種數(shù)據(jù)、何種容量，所有數(shù)據(jù)都無限保存是不實(shí)際的，成本方面也不經(jīng)濟(jì)。

Q3：實(shí)時(shí)計(jì)算場(chǎng)景有必要提升到秒級(jí)延遲嗎？

A3：延遲級(jí)別由業(yè)務(wù)場(chǎng)景決定。比如天級(jí)別的運(yùn)營報(bào)表本身具有意義，如果提升到秒級(jí)，數(shù)據(jù)量非常大，就失去了統(tǒng)計(jì)意義。但如果是前文分享的廣告案例，數(shù)據(jù)越實(shí)時(shí)，準(zhǔn)確性越高，業(yè)務(wù)上的效果就越好。

Q4：RCP和Apache DolphinScheduler一樣嗎？

A4：不一樣，DolphinScheduler具有工作流調(diào)度的功能，RCP主要負(fù)責(zé)實(shí)時(shí)計(jì)算的流任務(wù)管理。

劉騁昺

•  愛奇藝 大數(shù)據(jù)團(tuán)隊(duì) 高級(jí)經(jīng)理
• 愛奇藝大數(shù)據(jù)計(jì)算組負(fù)責(zé)人，負(fù)責(zé)愛奇藝大數(shù)據(jù)計(jì)算服務(wù)、實(shí)時(shí)計(jì)算平臺(tái)、實(shí)時(shí)分析平臺(tái)、機(jī)器學(xué)習(xí)平臺(tái)等系統(tǒng)的建設(shè)工作，擁有豐富的大數(shù)據(jù)領(lǐng)域?qū)崙?zhàn)經(jīng)驗(yàn)。