實(shí)時(shí)數(shù)倉(cāng) | 網(wǎng)易云音樂(lè)實(shí)時(shí)數(shù)倉(cāng)2.0進(jìn)階之路

云音樂(lè)從2018年開(kāi)始搭建實(shí)時(shí)計(jì)算平臺(tái)，經(jīng)過(guò)兩年的發(fā)展實(shí)時(shí)計(jì)算已經(jīng)滲透到云音樂(lè)的各個(gè)業(yè)務(wù)當(dāng)中：

• 運(yùn)營(yíng)需要實(shí)時(shí)的統(tǒng)計(jì)報(bào)表做精細(xì)化的運(yùn)營(yíng)

• 算法同學(xué)需要實(shí)時(shí)的特征數(shù)據(jù)來(lái)提升推薦效果、需要實(shí)時(shí)的AB數(shù)據(jù)來(lái)降低試錯(cuò)成本

• 搜索需要實(shí)時(shí)索引數(shù)據(jù)來(lái)提升線上搜索的效果

• 業(yè)務(wù)開(kāi)發(fā)需要實(shí)時(shí)的監(jiān)控?cái)?shù)據(jù)來(lái)第一時(shí)間響應(yīng)線上的問(wèn)題

  ......

發(fā)展到今天我們已經(jīng)擁有開(kāi)發(fā)者160+，線上任務(wù)運(yùn)行任務(wù)數(shù)500+，單Kafka的峰值流量超過(guò)了400W+QPS，實(shí)時(shí)集群機(jī)器數(shù)量130+，而這還僅僅是開(kāi)始，單單2020年Q1這段時(shí)間我們整體的機(jī)器以及任務(wù)的增量都超過(guò)了100%。

業(yè)務(wù)的飛速發(fā)展，開(kāi)發(fā)水平的層次不齊，整體也暴露出了越來(lái)越多的問(wèn)題，平臺(tái)運(yùn)維的壓力也越來(lái)越大，如何服務(wù)好業(yè)務(wù)，提升開(kāi)發(fā)者的開(kāi)發(fā)效率；如何做好運(yùn)維工作，保障整體平臺(tái)的穩(wěn)定性，都是擺在我們面前的問(wèn)題。

2018年我們發(fā)布了第一版的實(shí)時(shí)計(jì)算平臺(tái)，整體基于Flink1.7版本開(kāi)發(fā)，提供SQL和JAR包兩種開(kāi)發(fā)方式。JAR包任務(wù)比較簡(jiǎn)單，用戶(hù)基于官方依賴(lài)庫(kù)開(kāi)發(fā)，上傳即可；SQL方式我們采用了Antrl自定義了自己的一套SQL語(yǔ)法，實(shí)現(xiàn)了DDL、SET、CREATE FUNCTION、DIM JOIN 等語(yǔ)義，實(shí)現(xiàn)了流表的創(chuàng)建、屬性設(shè)置、函數(shù)創(chuàng)建、以及維表JOIN等功能，然后集成FLINK1.7本身SQL的能力實(shí)現(xiàn)了一套SQL的開(kāi)發(fā)能力，其整體架構(gòu)如下：

?

和目前業(yè)界很多的流計(jì)算平臺(tái)一樣，用戶(hù)在使用SQL開(kāi)發(fā)時(shí)需要先定義DDL，然后使用SQL定義開(kāi)發(fā)邏輯，具體可以參考阿里云或者sloth目前公開(kāi)出來(lái)的實(shí)時(shí)計(jì)算平臺(tái)的版本，基本使用方式差不多。在業(yè)務(wù)初期這套架構(gòu)很好的滿足了我們業(yè)務(wù)的發(fā)展，很好的解決了很多問(wèn)題，但是隨著業(yè)務(wù)越來(lái)越大，我們也發(fā)現(xiàn)了很多問(wèn)題：

數(shù)據(jù)理解問(wèn)題

在每次開(kāi)發(fā)任務(wù)前，開(kāi)發(fā)都需要了解以下幾個(gè)問(wèn)題

• 需要的數(shù)據(jù)從哪里來(lái)？需要去了解數(shù)據(jù)源信息，了解數(shù)據(jù)源的讀取方式
  

• 需要了解數(shù)據(jù)格式是什么，如何去解析

• 了解數(shù)據(jù)的具體的schema以及業(yè)務(wù)意義

整個(gè)開(kāi)發(fā)流程復(fù)雜冗長(zhǎng)，缺乏統(tǒng)一的一個(gè)地方讓我們查詢(xún)這些信息，而且整個(gè)過(guò)程缺乏復(fù)用的能力，每個(gè)人如果需要讀取相同的數(shù)據(jù)都要重復(fù)的去了解一遍，成本太高而且容易出錯(cuò)。

任務(wù)管控問(wèn)題

• 老版本的開(kāi)發(fā)基本還是基于數(shù)據(jù)源的開(kāi)發(fā)方式，整體難以管控。如果數(shù)據(jù)源做了遷移，需要去一個(gè)個(gè)的找到相關(guān)的任務(wù)然后一個(gè)個(gè)的修改，整體復(fù)雜冗長(zhǎng)很難運(yùn)維；

• 血緣追蹤問(wèn)題，老版本的任務(wù)缺乏統(tǒng)一的數(shù)據(jù)源管理，導(dǎo)致很難做血緣的track和追蹤，如果一個(gè)源頭數(shù)據(jù)發(fā)生變化很難track下游任務(wù)評(píng)估其影響范圍；

• Jar包任務(wù)缺乏統(tǒng)一的風(fēng)格、版本的控制，導(dǎo)致平臺(tái)需要支持不同版本的FLINK版本，平臺(tái)的開(kāi)發(fā)成本越來(lái)越高，需要適配不同版本的不同配置，整體的運(yùn)維成本也隨之增加。整體對(duì)JAR包任務(wù)的管控基本為0我們不清楚用戶(hù)的任務(wù)干什么的，屬于什么級(jí)別的任務(wù)，血緣追蹤也變得非常困難，整體平臺(tái)的能力和一些最佳實(shí)踐也很難快速的賦能給用戶(hù)。

SQL易用性問(wèn)題

老版本的SQL實(shí)現(xiàn)是在原生SQL的基礎(chǔ)上做了一層Antlr包裝，自定義了一些我們自己的語(yǔ)法，由于經(jīng)驗(yàn)的缺乏以及當(dāng)時(shí)Flink的SQL還處于初期的問(wèn)題，整體設(shè)計(jì)上存在一些問(wèn)題，使用也有很多限制。如一些語(yǔ)法和業(yè)界普遍使用的SQL語(yǔ)法存在很多出入，唯表JOIN的實(shí)現(xiàn)存在很多限制等，整體上和官方語(yǔ)法有一些出入，學(xué)習(xí)成本也偏高。

隨著我們整體平臺(tái)服務(wù)生態(tài)完善，以及FLINK本身的發(fā)展，我們?cè)?019年底開(kāi)始做底層技術(shù)的改造，開(kāi)發(fā)有元數(shù)據(jù)體系的SQL開(kāi)發(fā)框架，希望解決上面提到幾個(gè)問(wèn)題，提升平臺(tái)的穩(wěn)定性。

通過(guò)統(tǒng)一的元數(shù)據(jù)管理來(lái)解決數(shù)據(jù)理解問(wèn)題，整體架構(gòu)如下：

?

MetaHub元數(shù)數(shù)據(jù)中心是我們服務(wù)生態(tài)里面統(tǒng)一的元數(shù)據(jù)服務(wù)，其整體架構(gòu)如下：

?

元數(shù)據(jù)中心在我們數(shù)據(jù)平臺(tái)的整個(gè)服務(wù)生態(tài)中功能類(lèi)似于Apache Altas，但是功能比Altas更加完善，詳細(xì)在這里不做太多詳細(xì)的介紹，簡(jiǎn)單介紹下源數(shù)據(jù)中心關(guān)鍵的幾個(gè)功能：

• 采用元數(shù)據(jù)下推的方式管理Oracle、Kudu 、Hive、ES、HBase 等等自身?yè)碛性獢?shù)據(jù)系統(tǒng)的數(shù)據(jù)源，且可以通過(guò)插件的方式擴(kuò)展管理幾乎所有我們使用的數(shù)據(jù)源；

• 擁有獨(dú)立的元數(shù)據(jù)系統(tǒng)Nest用來(lái)管理Kafka、Nydus等沒(méi)有元數(shù)據(jù)體系的存儲(chǔ)中間件，Nest的整體實(shí)現(xiàn)參考了HIVE的元數(shù)據(jù)系統(tǒng)；

• 擁有統(tǒng)一類(lèi)型系統(tǒng)，在擴(kuò)展數(shù)據(jù)源開(kāi)發(fā)插件時(shí)，插件中需要實(shí)現(xiàn)相應(yīng)數(shù)據(jù)源到元數(shù)據(jù)中心類(lèi)型系統(tǒng)的映射方法；外部系統(tǒng)在接入元數(shù)據(jù)中心系統(tǒng)時(shí)不需要考慮不同數(shù)據(jù)源的類(lèi)型問(wèn)題，只需要關(guān)心元數(shù)據(jù)中心本身的一套類(lèi)型系統(tǒng)即可；

• 擁有元數(shù)據(jù)檢索功能，方便查找想要的數(shù)據(jù)；

• 擁有需要血緣追蹤模塊，只需要向元數(shù)據(jù)中心上傳任務(wù)的上下游關(guān)系，就可以快速的活動(dòng)整個(gè)的血緣信息。

Flink從1.9開(kāi)始提供了全新的Catalog API，外部的Calalog可以很方便的插拔式的接入Flink，我們希望基于這套全新的CatalogAPI，將Flink和元數(shù)據(jù)中心打通，將元數(shù)據(jù)中心作為Flink SQL的元數(shù)據(jù)管理系統(tǒng)，為了實(shí)現(xiàn)這一想法我們實(shí)現(xiàn)了自己MaginaCatalog，我們主要做了以下幾個(gè)事情：

類(lèi)型系統(tǒng)轉(zhuǎn)換：因?yàn)樵獢?shù)據(jù)中心有一套自己的類(lèi)型系統(tǒng)，我們只需要實(shí)現(xiàn)元數(shù)據(jù)中心這套類(lèi)型系統(tǒng)到Flink類(lèi)型系統(tǒng)的轉(zhuǎn)換就可以實(shí)現(xiàn)任何元數(shù)據(jù)中心支持?jǐn)?shù)據(jù)源的Table的Schema到Flink Table Schema的轉(zhuǎn)換。

?

整體大部分的工作元數(shù)據(jù)中心都幫我們Cover掉了，實(shí)現(xiàn)比較簡(jiǎn)單。

數(shù)據(jù)源屬性轉(zhuǎn)換：連接信息屬性轉(zhuǎn)換，F(xiàn)link每一種Connector都有自己的一套properties系統(tǒng)和元數(shù)據(jù)中心中的連接的屬性信息基本都不相同。如mysql的連接地址在元數(shù)據(jù)中心的key叫url但是在Flink中JDBC的Connector的連接地址的屬性叫connector.url，我們需要做一些key的轉(zhuǎn)換，這塊工作相對(duì)繁碎一點(diǎn)，但是實(shí)現(xiàn)起來(lái)相對(duì)簡(jiǎn)單。

Table Connector：目前官方提供的Table Connector的還不是很完善，我們根據(jù)我們需求完善了HBase、Kudu、Nydus、Redis、DDB等Connector。

完成以上工作以后，整體就基本實(shí)現(xiàn)我們的想法；在一個(gè)Flink任務(wù)的開(kāi)發(fā)中基本涉及的數(shù)據(jù)源主要有三類(lèi)：

• 流式數(shù)據(jù)：來(lái)自kafka或者 nydus，可以作為源端和目標(biāo)端

• 維表JOIN數(shù)據(jù)：來(lái)自HBase、Redis、JDBC等，這個(gè)取決于我們自己實(shí)現(xiàn)了哪些

• 落地?cái)?shù)據(jù)源：一般為Mysql、Hbase、Kudu、JDBC等，在流處理模式下一般作為目標(biāo)端

對(duì)于流式數(shù)據(jù)，我們使用元數(shù)據(jù)中心自帶的元數(shù)據(jù)系統(tǒng)Nest登記管理；對(duì)于維表以及落地?cái)?shù)據(jù)源等落地?cái)?shù)據(jù)源，可以直接通過(guò)元數(shù)據(jù)中心獲取庫(kù)表Schema信息，無(wú)需額外的Schema登記，只需要一次性登記下數(shù)據(jù)源連接信息即可，整體對(duì)應(yīng)我們系統(tǒng)中數(shù)倉(cāng)模塊的元數(shù)據(jù)管理、數(shù)據(jù)源登記兩個(gè)頁(yè)面。

數(shù)據(jù)訪問(wèn)：利用flink catalog提供的API，我們通過(guò)catalog、db、table三個(gè)元素來(lái)定位一張表，catalog相當(dāng)于數(shù)據(jù)源的標(biāo)識(shí)，db和table和其它數(shù)據(jù)庫(kù)系統(tǒng)類(lèi)似表示庫(kù)和表，如我們?cè)L問(wèn)截圖hive_da_music數(shù)據(jù)源的music_dwd庫(kù)中的user_play_fact表，只需要使用hive_da_music.music_dwd.user_play_fact訪問(wèn)即可，其中music_dwd.user_play_fact不需要額外登記直接使用的就是hive_da_music這個(gè)hive庫(kù)中的DB和Table。

DEMO：

SET?'ods_rtrs_ab_log.connections.group.id'?=?'hzwangeli2013_ab_rtrs_user_group_hour_rt_v2';??SET?'magina.sql.state.idleTime'?=?'14h';??SET?'table.exec.mini-batch.enabled'?=?'true';??SET?'table.exec.mini-batch.allow-latency'?=?'60s';??SET?'table.exec.mini-batch.size'?=?'100000';??CREATE?VIEW?rtrs_log?AS??SELECT??????from_unixtime(`timestamp`,?'YYYY-MM-dd')?as?dt,??????'all'?as?os,??????sceneid,??????parent_exp,??????cast(from_unixtime(`timestamp`,?'HH')?as?int)?as?`hour`,??????`timestamp`?log_time,??????`exp`,??????cast(exp_type?as?int)?exp_type,??????abs(cast(userid?as?bigint))?userid,??????'0'?trace_id???FROM?iplay_ods.ods_rtrs_ab_log???WHERE?`timestamp`?<=?UNIX_TIMESTAMP()?+?10??AND???`timestamp`?>=?UNIX_TIMESTAMP()?-?12?*?3600??AND?abs(cast(userid?as?bigint))?>?0??AND?sceneid?IS?NOT?NULL???AND?parent_exp?IS?NOT?NULL???AND?`exp`?IS?NOT?NULL???AND?exp_type?IS?NOT?NULL;??INSERT?INTO?`music_kudu_online`.`music_kudu_internal`.`ab_rtrs_user_group_hour_rt_v2`??SELECT???????dt,???????os,???????sceneid,??????parent_exp,??????`hour`,??????`exp`,??????exp_type,??????userid,??????trace_id,??????count(1)?pv,??????min(log_time)?min_time,??????max(log_time)?max_time???FROM?rtrs_log  GROUP BY dt, os, sceneid, parent_exp, `hour`, `exp`, exp_type, userid, trace_id

??

• 通過(guò)SET語(yǔ)句配置ods_rtrs_ab_log流表的groupid、狀態(tài)的過(guò)期時(shí)間

• iplay_ods.ods_rtrs_ab_log：一張流表，進(jìn)行了額外的登記，沒(méi)有指定catalog，是因?yàn)榱鞅淼腃atalog是默認(rèn)的無(wú)需額外的指定

• music_kudu_online.music_kudu_internal.ab_rtrs_user_group_hour_rt_v2: 是一張kudu表只需要登記kudu的數(shù)據(jù)源地址即可，music_kudu_internal.ab_rtrs_user_group_hour_rt_v2是kudu的表名，因?yàn)閗udu本身沒(méi)有庫(kù)概念，這里我們做了一點(diǎn)特殊處理，在此不做額外的贅述

• 和普通數(shù)據(jù)庫(kù)系統(tǒng)一樣，只需要簡(jiǎn)單一個(gè)INSERT INTO語(yǔ)句就完成了流數(shù)據(jù)往Kudu的寫(xiě)入操作

對(duì)于JAR包任務(wù)，我們還提供了一個(gè)實(shí)時(shí)數(shù)倉(cāng)的SDK，用戶(hù)通過(guò)這套SDK可以通過(guò)SQL加DataStreamAPI的方式進(jìn)行混合編程，大大降低了開(kāi)發(fā)成本，用戶(hù)不需要關(guān)心數(shù)據(jù)源的真實(shí)的連接信息，只需要關(guān)注catalog和庫(kù)表即可，如：

?public?class?CodeDemo??extends?MaginaBase?{??????protected?void?run(String?args[])?throws?Exception?{??????????Table?table?=?session.sql("SET?'ods_music_ua_queue_3.connections.group.id'?=?'skd_jww_test';\\n"?+??????????????????"select?os?from?magina_dw_online.music_ods.ods_music_ua_queue_3?where?os?=?'pc';").get();??????????DataStream?result?=?toDataStream(table);????????result.print()??????}  }

讀取數(shù)據(jù)非常的簡(jiǎn)單明了無(wú)需關(guān)注太多的數(shù)據(jù)源連接信息、類(lèi)型信息，只需要繼承我們的基礎(chǔ)類(lèi)即可。

前端優(yōu)化

元數(shù)據(jù)的搜索和代碼生成功能，方便用戶(hù)查找的想要外部表，并生成相應(yīng)的SQL代碼

?

debug功能：我們參考了flink on zeppelin的實(shí)現(xiàn)，實(shí)現(xiàn)了一套自己的debug服務(wù)，用戶(hù)可以通過(guò)debug獲取結(jié)果。

?

為了方便一些特殊數(shù)據(jù)端的寫(xiě)入，以及方便我們后續(xù)跳出SQL規(guī)范枷鎖，做一些額外的功能我們還開(kāi)發(fā)了SINK輸出組件，用戶(hù)只需要簡(jiǎn)單的勾勾選選就可以完成數(shù)據(jù)輸出的開(kāi)發(fā)工作。

?

如HDFS輸出組件我們開(kāi)發(fā)了過(guò)期時(shí)間的功能做歸檔文件的過(guò)期自動(dòng)清理，這個(gè)功能純SQL是沒(méi)有辦法優(yōu)雅的實(shí)現(xiàn)的，除了常用的數(shù)據(jù)源以外我們還實(shí)現(xiàn)了我們音樂(lè)自研的一些系統(tǒng)的SINK，如監(jiān)控系統(tǒng)Pylon、索引系統(tǒng)NSearch等這些都可以方便在SINK組件中實(shí)現(xiàn)。

總結(jié)

我們?cè)倩貋?lái)看下上面提到的幾個(gè)問(wèn)題。

1.數(shù)據(jù)理解問(wèn)題

我們通過(guò)統(tǒng)計(jì)的元數(shù)據(jù)登記解決數(shù)據(jù)理解問(wèn)題，用戶(hù)無(wú)需關(guān)注數(shù)據(jù)源信息、數(shù)據(jù)如何序列化、有哪些字段，只需要去數(shù)倉(cāng)管理中查找自己想要的表即可。

2.任務(wù)管控問(wèn)題

• 通過(guò)登記的方式管理所有的數(shù)據(jù)源，如果數(shù)據(jù)源發(fā)送變成只需要在統(tǒng)一的地方修改，然后重啟下游任務(wù)即可，用戶(hù)的代碼不需要任何修改；

• 通過(guò)SQL管理所有的SQL任務(wù)，通過(guò)SDK的方式統(tǒng)一管控所有JAR的任務(wù)，統(tǒng)一版本和開(kāi)發(fā)風(fēng)格；

• 管控了所有的任務(wù)的開(kāi)發(fā)方式，我們正在開(kāi)發(fā)血緣模塊來(lái)解決任務(wù)血緣追蹤的問(wèn)題，實(shí)現(xiàn)這一點(diǎn)也非常的簡(jiǎn)單只需要將任務(wù)的血緣信息上報(bào)到源數(shù)據(jù)中心即可。

整體任務(wù)管控這一塊我們還有很多的規(guī)劃，整體可以參考我們后續(xù)規(guī)劃的章節(jié)。

3.SQL易用性問(wèn)題

因?yàn)檎w基于Flink本身的Catalog API實(shí)現(xiàn)，整體SQL規(guī)范和官方版本相同，無(wú)需額外的學(xué)習(xí)成本，用戶(hù)在外網(wǎng)上也能找到很多相關(guān)的資料，另外我們還基于官方的維表JOIN的API實(shí)現(xiàn)了完美版本的唯表JOIN功能，易用性上有了大大的提升。

1.數(shù)據(jù)血緣功能

通過(guò)SQL解析的方式，上報(bào)血緣信息到源數(shù)據(jù)中心，實(shí)現(xiàn)任務(wù)以及流表血緣追蹤。

2.通過(guò)任務(wù)分類(lèi)實(shí)現(xiàn)運(yùn)維部署的分級(jí)

• 我們希望將數(shù)據(jù)源歸類(lèi)成線上、統(tǒng)計(jì)、測(cè)試等類(lèi)別,然后通過(guò)血緣信息自動(dòng)判定任務(wù)是線上任務(wù)、統(tǒng)計(jì)任務(wù)、還是測(cè)試任務(wù)；

• 交互上用戶(hù)主動(dòng)給任務(wù)打標(biāo)。

通過(guò)任務(wù)類(lèi)別的不同我們可以針對(duì)不同級(jí)別的任務(wù)做不同級(jí)別的管理和硬件規(guī)劃,如線上、統(tǒng)計(jì)、測(cè)試任務(wù)集群物理級(jí)別的隔離。

3.源端目標(biāo)端的統(tǒng)一管理

目前這點(diǎn)我們主要focus在kafka的監(jiān)控上我們希望通過(guò)整合kakfa的監(jiān)控信息、任務(wù)的血緣信息、以及任務(wù)重要級(jí)別做一個(gè)全鏈路的監(jiān)控，當(dāng)kafka出現(xiàn)問(wèn)題時(shí)我們能夠第一時(shí)間定位到是誰(shuí)是那個(gè)任務(wù)導(dǎo)致kakfa的異常，通過(guò)任務(wù)類(lèi)別的不同我們還可以自動(dòng)化的選擇主集群還是鏡像集群做到kafka集群的物理隔離，來(lái)防止統(tǒng)計(jì)任務(wù)對(duì)線上任務(wù)產(chǎn)生影響。

4.特征snaohot以及實(shí)時(shí)模型訓(xùn)練平臺(tái)級(jí)別的支持

業(yè)務(wù)相關(guān)，后續(xù)有了落地再給大家介紹。

5.批流一體的整合

可以關(guān)注下上一篇文章?[2021年網(wǎng)易云音樂(lè)實(shí)時(shí)計(jì)算平臺(tái)發(fā)展和挑戰(zhàn)]，有相關(guān)內(nèi)容介紹。