摘要：本文由知乎技術(shù)平臺負(fù)責(zé)人孫曉光分享，主要介紹知乎 Flink 數(shù)據(jù)集成平臺建設(shè)實(shí)踐。內(nèi)容如下：

• 業(yè)務(wù)場景

• 歷史設(shè)計(jì)

• 全面轉(zhuǎn)向 Flink 后的設(shè)計(jì)

• 未來 Flink 應(yīng)用場景的規(guī)劃

01

業(yè)務(wù)場景

很高興和大家分享近期知乎以 Flink 為基礎(chǔ)，重構(gòu)上一代數(shù)據(jù)集成平臺過程中的一些收獲。數(shù)據(jù)集成平臺作為連接各種異構(gòu)數(shù)據(jù)的紐帶，需要連接多種多樣的存儲系統(tǒng)。而不同的技術(shù)棧和不同的業(yè)務(wù)場景會對數(shù)據(jù)集成系統(tǒng)提出不同的設(shè)計(jì)要求。

?

我們首先來看一下在知乎內(nèi)部數(shù)據(jù)集成的業(yè)務(wù)場景。同許多互聯(lián)網(wǎng)公司相似，過去知乎的在線存儲系統(tǒng)主要以 MySQL 和 Redis 為主，同時(shí)對于部分?jǐn)?shù)據(jù)量級較大的業(yè)務(wù)也使用了 HBase。近年來隨著技術(shù)的演進(jìn)，我們開始了從 MySQL 向 TiDB 的遷移。與此類似，我們也開始將 HBase 向基于 TiKV 技術(shù)棧研發(fā)的 Zetta 演進(jìn)。在離線存儲方面絕大多數(shù)的場景則是以 Hive 表來支撐的。

?

從在線存儲到離線存儲，期間有著非常強(qiáng)的數(shù)據(jù)同步需求。除此以外也存在著大量的流式數(shù)據(jù)，比如消息系統(tǒng)中的數(shù)據(jù)，我們也希望它能夠同各種在線或離線存儲系統(tǒng)打通。過去知乎主要使用 Kafka 支撐流式數(shù)據(jù)，近期也開始引入 Pulsar。這兩套消息系統(tǒng)同存儲系統(tǒng)之間的數(shù)據(jù)交換存在著較強(qiáng)的需求。

?

?

在知乎的業(yè)務(wù)場景和當(dāng)前發(fā)展?fàn)顟B(tài)下，數(shù)據(jù)集成工作在技術(shù)和流程管理上都存在著一些挑戰(zhàn)。

?

• 首先從技術(shù)角度看，數(shù)據(jù)源多樣化會對數(shù)據(jù)集成系統(tǒng)的連接擴(kuò)展能力提出較高的要求。而且下一代的存儲系統(tǒng)在給業(yè)務(wù)帶來更強(qiáng)能力的同時(shí)也釋放了業(yè)務(wù)的壓力，進(jìn)而促使了數(shù)據(jù)量的加速膨脹。數(shù)據(jù)量級上的快速增長對數(shù)據(jù)集成平臺的吞吐和實(shí)時(shí)性都提出了更高的要求。當(dāng)然作為數(shù)據(jù)相關(guān)的基礎(chǔ)系統(tǒng)，數(shù)據(jù)準(zhǔn)確性則是最基礎(chǔ)的要求，這塊我們也必須把它做好。

  ?

• 另外從流程管理角度看，我們需要理解并整合散落在不同業(yè)務(wù)團(tuán)隊(duì)的數(shù)據(jù)，做好管理并確保數(shù)據(jù)訪問的安全，所以整個(gè)數(shù)據(jù)整合的流程是相對復(fù)雜的。雖然平臺化能夠?qū)?fù)雜的流程自動化起來，但數(shù)據(jù)集成工作所固有的高成本并不能完全以平臺化的方式消除。因此盡最大的可能提升流程的可復(fù)用性和可管理性也是數(shù)據(jù)集成系統(tǒng)需要持續(xù)應(yīng)對的挑戰(zhàn)。

?

?

基于這兩個(gè)方向上的挑戰(zhàn)，我們對數(shù)據(jù)集成平臺的設(shè)計(jì)目標(biāo)進(jìn)行了規(guī)劃。

?

• 從技術(shù)方向看，我們需要支持知乎已經(jīng)投入使用和將來要推廣使用的多種存儲系統(tǒng)，具備將這些系統(tǒng)中多樣化的數(shù)據(jù)進(jìn)行集成的能力。此外我們還需要在滿足高吞吐，低調(diào)度時(shí)延的前提下保障數(shù)據(jù)集成的可靠性和準(zhǔn)確性。

  ?

• 從流程方面看，可以通過整合各種內(nèi)部存儲系統(tǒng)的元數(shù)據(jù)以及調(diào)度系統(tǒng)，復(fù)用現(xiàn)有系統(tǒng)基礎(chǔ)設(shè)施的能力，達(dá)到簡化數(shù)據(jù)接入流程，降低用戶接入成本的目的。我們還希望能夠以平臺化的方式為用戶提供自助滿足數(shù)據(jù)需求的手段，從而提升數(shù)據(jù)集成工作的整體效率。

  ?

• 從提升任務(wù)可管理性的角度看，我們還要維護(hù)好數(shù)據(jù)的血緣關(guān)系。讓業(yè)務(wù)更好的去度量數(shù)據(jù)產(chǎn)出之間的關(guān)系，更有效的評估數(shù)據(jù)產(chǎn)出的業(yè)務(wù)價(jià)值，避免低質(zhì)量和重復(fù)性的數(shù)據(jù)集成工作。最后我們需要對所有任務(wù)提供系統(tǒng)化的監(jiān)控和報(bào)警能力來保障數(shù)據(jù)產(chǎn)出的穩(wěn)定性。

?

02

歷史設(shè)計(jì)

在知乎的第一代數(shù)據(jù)集成平臺成型前，大量的任務(wù)散落在各個(gè)業(yè)務(wù)方自己維護(hù)的 crontab 或者自行搭建的各種調(diào)度系統(tǒng)中。在這樣的無管理狀態(tài)下，各項(xiàng)集成任務(wù)的可靠性和數(shù)據(jù)質(zhì)量都很難得到有效的保障。因此在這個(gè)階段我們要最迫切解決的是管理上的問題，讓數(shù)據(jù)集成的流程可管理可監(jiān)控。

?

因此，我們整合了各種存儲系統(tǒng)的元數(shù)據(jù)系統(tǒng)，讓大家可以在統(tǒng)一的地方看到公司所有的數(shù)據(jù)資產(chǎn)。然后在調(diào)度中心統(tǒng)一管理這些數(shù)據(jù)的同步任務(wù)，由調(diào)度中心負(fù)責(zé)任務(wù)的依賴管理。同時(shí)調(diào)度中心對任務(wù)的關(guān)鍵指標(biāo)進(jìn)行監(jiān)控并提供異常告警能力。在這個(gè)階段我們沿用了從前大家廣泛使用的 Sqoop 來實(shí)現(xiàn) MySQL 和 Hive 之間數(shù)據(jù)的同步。且在平臺建設(shè)后期，隨著流數(shù)據(jù)同步需求的出現(xiàn)，我們又引入了 Flink 來同步 Kafka 數(shù)據(jù)到 HDFS。

?

?

在建設(shè)初代集成平臺時(shí)我們做過一次技術(shù)選型的選擇，是繼續(xù)使用已經(jīng)得到廣泛驗(yàn)證的 Sqoop 還是遷移到其它可選的技術(shù)方案。同 Sqoop 相比，阿里開源的 DataX 是這個(gè)領(lǐng)域一個(gè)非常有競爭力的對手。如果把這兩個(gè)產(chǎn)品進(jìn)行橫向?qū)Ρ?#xff0c;可以發(fā)現(xiàn)他們在不同的方面互相有對方所不具備的優(yōu)勢。

?

• 比如 Sqoop 在系統(tǒng)規(guī)模上具備 MapReduce 級別的擴(kuò)展性和原生的 Hive 支持。但 Sqoop 又有數(shù)據(jù)源支持不豐富，缺乏一些重要功能特性的缺點(diǎn)。

  ?

• 而 DataX 提供了非常豐富的數(shù)據(jù)源支持，內(nèi)置了數(shù)據(jù)集成系統(tǒng)非常重要的限速能力，還有它的良好設(shè)計(jì)所帶來的易于定制和擴(kuò)展的能力。但它也存在無集群資源管理支持和欠缺 Hive Catalog 原生支持的缺陷。

?

在當(dāng)時(shí)的狀態(tài)下這兩個(gè)產(chǎn)品相互比較起來沒有一款產(chǎn)品具有絕對的優(yōu)勢。所以我們選擇了繼續(xù)使用 Sqoop，而維持使用 Sqoop 在驗(yàn)證環(huán)節(jié)上也為我們節(jié)約了許多投入，所以第一代的數(shù)據(jù)集成平臺在非常短的時(shí)間內(nèi)就完成了開發(fā)和驗(yàn)證并完成上線。

?

?

隨著初代數(shù)據(jù)集成平臺的上線和成熟，它很好的支撐了公司的數(shù)據(jù)集成業(yè)務(wù)需求并獲得了顯著的收益。到目前為止平臺上一共有大約 4000 個(gè)任務(wù)，每天運(yùn)行超過 6000 個(gè)任務(wù)實(shí)例，同步大約 82 億條共計(jì) 124TB 的數(shù)據(jù)。

?

在平臺的幫助下，數(shù)據(jù)接入流程得到了極大的簡化，為用戶提供了自助解決數(shù)據(jù)集成需求的能力。并且，平臺在關(guān)鍵的流程節(jié)點(diǎn)上能夠輔以必要的規(guī)范約束和安全審查，在提升了管理水平的同時(shí)，整體的安全性和數(shù)據(jù)質(zhì)量也得到了顯著的提升。

?

借助于 Yarn 和 K8s 的彈性能力，集成任務(wù)的規(guī)模擴(kuò)展能力也有了很大的提升。當(dāng)然，作為解決從 0 到 1 問題的第一代系統(tǒng)，也必然會伴隨著一系列問題。比如：

?

• Sqoop 的 MapReduce 模式所固有的高調(diào)度時(shí)延問題

?

• 業(yè)務(wù)數(shù)據(jù)分布不均所導(dǎo)致的數(shù)據(jù)傾斜問題

?

• 社區(qū)不活躍導(dǎo)致部分 Issue 長期無法得到解決的問題

?

• Sqoop 代碼設(shè)計(jì)不理想導(dǎo)致的可擴(kuò)展性和可管理性弱的問題。

?

?

03

轉(zhuǎn)向 Flink

與 Sqoop 相對的，是用于支持 Kafka 消息到 HDFS 數(shù)據(jù)集成任務(wù)的 Flink，它以優(yōu)秀的可靠性和靈活的可定制性獲得了大家更多的信任。基于流式數(shù)據(jù)集成任務(wù)為 Flink 建立的信心，我們開始嘗試全面轉(zhuǎn)向 Flink 來建設(shè)下一代的數(shù)據(jù)集成平臺。

?

雖然 Flink 是本次平臺演進(jìn)中的最佳候選，我們還是基于當(dāng)時(shí)的情況對市面上可選的技術(shù)方案再次進(jìn)行了調(diào)研。這次我們將 Apache NIFI 項(xiàng)目和 Flink 進(jìn)行了多方面的比較，從功能角度看：

?

• Apache NIFI 非常強(qiáng)大且完全覆蓋了我們當(dāng)前的數(shù)據(jù)集成需求。但是恰恰因?yàn)樗δ苓^于強(qiáng)大并且自成體系，所以也帶來了較高的整合門檻。而且，無法利用現(xiàn)有 Yarn 和 K8s 資源池也會帶來額外的資源池建設(shè)和維護(hù)的成本。

  ?

• 相比之下， Flink 具有一個(gè)非常活躍和開放的社區(qū)，在立項(xiàng)時(shí)刻就已經(jīng)具備了非常豐富的數(shù)據(jù)源支持，可以預(yù)期在未來它的數(shù)據(jù)源覆蓋一定會更加全面。而且 Flink 作為一個(gè)通用計(jì)算引擎有著強(qiáng)大易用的 API 設(shè)計(jì)，在這個(gè)基礎(chǔ)上進(jìn)行二次開發(fā)非常容易，所以它在可擴(kuò)展性方面的優(yōu)勢也非常突出。

?

最后基于我們對批流一體目標(biāo)的認(rèn)同，未來在知乎完成大數(shù)據(jù)計(jì)算引擎技術(shù)棧的統(tǒng)一也是一個(gè)極具吸引力的目標(biāo)。

?

?

基于這些考量，在本輪迭代中我們選擇了全面使用 Flink 替代 Sqoop，基于 Flink 完整實(shí)現(xiàn)了之前 Sqoop 的功能并重新建設(shè)了全新的集成平臺。

?

如下圖所示，橙色部分是本輪迭代中發(fā)生了變化的部分。除了作為主角出現(xiàn)的 Flink 之外，在本輪迭代的過程中我們還開發(fā)了 TiDB、Redis 和 Zetta 三種存儲系統(tǒng)的數(shù)據(jù)集成功能。在消息系統(tǒng)這邊則直接從社區(qū)獲得了 Pulsar 的支持。在我們開始開發(fā)工作的時(shí)候，Flink 已經(jīng)演進(jìn)到了比較成熟的階段，對 Hive 內(nèi)建了原生的支持，整個(gè)遷移過程沒有遇到過多的技術(shù)困難，非常順暢。

?

?

Flink 的遷移為我們帶來了許多收益。

?

1.?首先從可維護(hù)性上看，相比 Sqoop 有了非常顯著的改善。如下圖所示，左邊是過去使用 Sqoop 時(shí)的任務(wù)定義，這里是一大堆非結(jié)構(gòu)化的容易出錯(cuò)的原始命令。而 Flink 則只需使用 SQL 定義一個(gè)源表和一個(gè)目標(biāo)表再配合寫入命令來定義任務(wù)。任務(wù)的可理解性、可調(diào)試性遠(yuǎn)好于從前，變成最終用戶也能夠理解的模式。很多問題不再需要平臺開發(fā)者配合排查，用戶就能夠自助的解決許多常見的任務(wù)異常。

?

?

2. 在性能角度方面，我們也有針對性的做了許多優(yōu)化。

?

2.1 調(diào)度策略

?

首先是調(diào)度策略上的優(yōu)化，在第一代集成平臺中我們只使用 Flink 同步流式數(shù)據(jù)，所以任務(wù)調(diào)度完全使用 Per Job。現(xiàn)在平臺同時(shí)支持了 Session 和 Per Job 的混合調(diào)度模式，于是，對于從消息系統(tǒng)接入數(shù)據(jù)的流式任務(wù)會繼續(xù)使用 Per-Job 模式運(yùn)行，而批同步的任務(wù)則采用 Session 模式復(fù)用集群從而避免集群啟動的耗時(shí)提升同步效率。

?

當(dāng)然，在這樣的場景中使用 Session 集群也存在著一系列的挑戰(zhàn)，比如工作負(fù)載隨著任務(wù)提交不停變化而帶來的資源需求變化問題。所以我們建設(shè)了自動的擴(kuò)縮容機(jī)制來幫助 Session 集群應(yīng)對變化的負(fù)載。除此以外，為了簡化計(jì)費(fèi)機(jī)制和隔離風(fēng)險(xiǎn)，我們還為不同的業(yè)務(wù)線創(chuàng)建了私有 Session 集群用于服務(wù)對應(yīng)業(yè)務(wù)線的數(shù)據(jù)集成任務(wù)。

?

?

2.2 數(shù)據(jù)庫

?

在關(guān)系數(shù)據(jù)庫方面，我們采用了常見的 JDBC 方式對 MySQL 進(jìn)行數(shù)據(jù)同步，但這種方式也會存在一些固有難以解決的問題。

?

• 比如因業(yè)務(wù)數(shù)據(jù)在主鍵維度上空間分布不均導(dǎo)致的數(shù)據(jù)傾斜問題。

  ?

• 再比如為了隔離在線離線工作負(fù)載所建設(shè)的專用同步從庫，所產(chǎn)生的資源浪費(fèi)和管理成本。

  ?

• 并且由于 MySQL 實(shí)例眾多規(guī)格不一，合理協(xié)調(diào)多個(gè)并發(fā)任務(wù)的實(shí)例和實(shí)例所在的主機(jī)，進(jìn)行合理的速度控制也非常困難。

?

?

相比之下，考慮到正在全面將數(shù)據(jù)從 MySQL 遷移到 TiDB 這一趨勢。我們開發(fā)了原生 TiDB 的 Flink connector 來充分利用 TiDB 架構(gòu)上的優(yōu)勢。

?

• 首先 region 級別的負(fù)載均衡策略能夠確保對于任何表結(jié)構(gòu)和任何的數(shù)據(jù)分布，同步任務(wù)都能夠以 region 為顆粒度進(jìn)行拆分避免數(shù)據(jù)傾斜問題。

  ?

• 其次通過設(shè)定副本放置策略，可以在離線數(shù)據(jù)中心對數(shù)據(jù)統(tǒng)一放置一個(gè) Follower 副本。進(jìn)而在保持原有目標(biāo)副本數(shù)量不變，無需額外資源成本的情況下，利用 Follower read 的能力隔離在線交易和數(shù)據(jù)抽取的負(fù)載。

  ?

• 最后我們還引入了分布式的數(shù)據(jù)提交方式提升了數(shù)據(jù)寫入的吞吐能力。

?

?

3. 最后是為知乎內(nèi)部廣泛使用的 Redis 提供數(shù)據(jù)集成的能力。Flink 社區(qū)已經(jīng)有一個(gè)? Redis connector，但它目前只具備寫入能力并且難以靈活定制寫入時(shí)所使用的? key。所以我們基于自身需求重新開發(fā)了一個(gè) Redis connector，同時(shí)支持以 Redis 作為 Source ?和 Sink。

?

同樣為了避免數(shù)據(jù)抽取過程影響在線交易，在數(shù)據(jù)讀取路徑上我們采用了 Redis 原生的 master/slave 機(jī)制獲取并解析 RDB 文件抽取數(shù)據(jù)，獲得了單實(shí)例約 150MB 每秒的數(shù)據(jù)抽取吞吐。而且得益于打通內(nèi)部存儲系統(tǒng)的元數(shù)據(jù)，我們不但能夠支持分片模式 Redis 集群的數(shù)據(jù)抽取，還可以只選擇每個(gè)分片的? slave 節(jié)點(diǎn)作為數(shù)據(jù)抽取源頭，避免抽取對 master 節(jié)點(diǎn)產(chǎn)生壓力。

?

?

這次全面轉(zhuǎn)向 Flink 的演進(jìn)，解決了很多上一代數(shù)據(jù)集成平臺的問題，獲得了非常顯著的收益。

?

從吞吐角度看，以 Flink 替代 MR 模式將整個(gè)調(diào)度的時(shí)延從分鐘級降低到了 10 秒左右。并且在同樣的數(shù)據(jù)量和同樣的 Flink 資源量情況下，TiDB 原生 connector 能夠比 JDBC 提升 4 倍的吞吐。

從功能角度看，新平臺不但能夠原生支持分庫分表的數(shù)據(jù)集成任務(wù)，還能夠以業(yè)務(wù)無關(guān)的方式避免數(shù)據(jù)傾斜的問題。

在數(shù)據(jù)源支持能力上，我們以非常低的成本獲得了 TiDB、Zetta、Redis 和 Pulsar 的支持。而且，隨著 Flink 的生態(tài)越來越完善，未來一定會有更多的開箱即用的 connector 供我們使用。

從成本上看，最后下線 MySQL 離線節(jié)點(diǎn)和統(tǒng)一使用 K8s 資源池所帶來的資源效率提升，在成本和管理角度看都使我們獲得了顯著的收益。

?

04

Flink 即未來

回過頭看，本次全面 Flink 化的演進(jìn)投入產(chǎn)出比非常高，這也進(jìn)一步增強(qiáng)了我們對? ?“Flink 即未來” 的的信心。目前在知乎內(nèi)部除了數(shù)據(jù)集成場景，Flink 在搜索 Query 的時(shí)效性分析、商業(yè)廣告點(diǎn)擊數(shù)據(jù)處理和關(guān)鍵業(yè)務(wù)指標(biāo)的實(shí)時(shí)數(shù)倉上也都有所應(yīng)用。

?

在未來我們希望能夠進(jìn)一步擴(kuò)展 Flink 在知乎的使用場景，建設(shè)更加全面的實(shí)時(shí)數(shù)倉、系統(tǒng)化的在線機(jī)器學(xué)習(xí)平臺。我們更希望批流一體的落地，讓報(bào)表類和 ETL 類的大型批任務(wù)也能夠在 Flink 平臺上落地。

?

基于知乎大數(shù)據(jù)系統(tǒng)建設(shè)的模式和總體資源投入的情況，未來將技術(shù)棧向 Flink 收攏是一個(gè)非常適合知乎的選擇。作為用戶我們非常期待能夠一同見證未來 Flink 批流一體目標(biāo)的達(dá)成。同時(shí)作為社區(qū)的成員，我們也希望能夠用自己的方式為這一目標(biāo)的達(dá)成貢獻(xiàn)一份力量。

總結(jié)

以上是生活随笔為你收集整理的知乎数据集成平台建设实践的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。