摘要：本文整理自戴爾科技集團(tuán)軟件工程師周煜敏在 Flink Forward Asia 2020 分享的議題《Pravega Flink Connector 的過(guò)去、現(xiàn)在和未來(lái)》，文章內(nèi)容為：

1. Pravega 以及 Pravega connector 簡(jiǎn)介
2. Pravega connector 的過(guò)去
3. 回顧 Flink 1.11 高階特性心得分享
4. 未來(lái)展望

一、Pravega 以及 Pravega connector 簡(jiǎn)介

Pravega 項(xiàng)目的名字來(lái)源于梵語(yǔ)，意思是 good speed。項(xiàng)目起源于 2016 年，基于 Apache V2 協(xié)議在 Github 上開(kāi)源，并且于 2020 年 11 月加入了 CNCF 的大家庭，成為了 CNCF 的 sandbox 項(xiàng)目。

Pravega 項(xiàng)目是為大規(guī)模數(shù)據(jù)流場(chǎng)景而設(shè)計(jì)的，彌補(bǔ)傳統(tǒng)消息隊(duì)列存儲(chǔ)短板的一個(gè)新的企業(yè)級(jí)存儲(chǔ)系統(tǒng)。它在保持對(duì)于流的無(wú)邊界、高性能的讀寫上，也增加了企業(yè)級(jí)的一些特性：例如彈性伸縮以及分層存儲(chǔ)，可以幫助企業(yè)用戶降低使用和維護(hù)的成本。同時(shí)我們也在存儲(chǔ)領(lǐng)域有著多年的技術(shù)沉淀，可以依托公司商用存儲(chǔ)產(chǎn)品為客戶提供持久化的存儲(chǔ)。

以上的架構(gòu)圖描述的是 Pravega 典型的讀寫場(chǎng)景，借此進(jìn)行 Pravega 術(shù)語(yǔ)介紹以幫助大家進(jìn)一步了解系統(tǒng)架構(gòu)。

• 中間部分是一個(gè) Pravega 的集群 ，它整體是以 stream 抽象的系統(tǒng)。stream 可以認(rèn)為是類比 Kafka 的 topic。同樣，Pravega 的 Segment 可以類比 Kafka 的 Partition，作為數(shù)據(jù)分區(qū)的概念，同時(shí)提供動(dòng)態(tài)伸縮的功能。

Segment 存儲(chǔ)二進(jìn)制數(shù)據(jù)數(shù)據(jù)流，并且根據(jù)數(shù)據(jù)流量的大小，發(fā)生 merge 或者 split 的操作，以釋放或者集中資源。此時(shí) Segment 會(huì)進(jìn)行 seal 操作禁止新數(shù)據(jù)寫入，然后由新建的 Segment 進(jìn)行新數(shù)據(jù)的接收。

• 圖片左側(cè)是數(shù)據(jù)寫入的場(chǎng)景，支持 append only 的寫入。用戶可以對(duì)于每一個(gè) event 指定 Routing key 來(lái)決定 Segment 的歸屬。這一點(diǎn)可以類比 Kafka Partitioner。單一的 Routing key 上的數(shù)據(jù)具有保序性，確保讀出的順序與寫入相同。
• 圖片右側(cè)是數(shù)據(jù)讀取的場(chǎng)景，多個(gè) reader 會(huì)有一個(gè) Reader Group 進(jìn)行管控。Reader Group 控制著 reader 之間的負(fù)載均衡的，來(lái)保證所有的 Segment 能在 reader 之間均勻分布。同時(shí)也提供 Checkpoint 機(jī)制形成一致的 stream 切分來(lái)保證數(shù)據(jù)的故障恢復(fù)。對(duì)于 "讀"，我們支持批和流兩種語(yǔ)義。對(duì)于流的場(chǎng)景，我們支持尾讀;對(duì)于批的場(chǎng)景，我們會(huì)更多的考慮高并發(fā)來(lái)達(dá)到高吞吐。

二、Pravega Flink connector 的過(guò)去

Pravega Flink connector 是 Pravega 最初支持的 connector，這也是因?yàn)?Pravega 與 Flink 的設(shè)計(jì)理念非常一致，都是以流為基礎(chǔ)的批流一體的系統(tǒng)，能夠組成存儲(chǔ)加計(jì)算的完整解決方案。

1. Pravega 發(fā)展歷程

• connector 從 2017 年開(kāi)始成為獨(dú)立的 Github 項(xiàng)目。2017 年，我們基于 Flink 1.3 版本進(jìn)行開(kāi)發(fā)，當(dāng)時(shí)有包括 Stephan Ewen 在內(nèi)的 Flink PMC 成員加入，合作構(gòu)建了最基礎(chǔ)的 Source / Sink function，支持最基礎(chǔ)的讀寫，同時(shí)也包括 Pravega Checkpoint 的集成，這點(diǎn)會(huì)在后面進(jìn)行介紹。
• 2018 年最重要的一個(gè)亮點(diǎn)功能就是端到端的精確一次性語(yǔ)義支持。當(dāng)時(shí)團(tuán)隊(duì)和 Flink 社區(qū)有非常多的討論，Pravega 首先支持了事務(wù)性寫客戶端的特性，社區(qū)在此基礎(chǔ)上合作，以 Sink function 為基礎(chǔ)，通過(guò)一套兩階段提交的語(yǔ)義實(shí)現(xiàn)了基于 checkpoint 的分布式事務(wù)功能。后來(lái)，F(xiàn)link 也進(jìn)一步抽象出了兩階段提交的 API，也就是為大家熟知的 TwoPhaseCommitSinkFunction 接口，并且也被 Kafka connector 采用。社區(qū)有博客來(lái)專門介紹這一接口，以及端到端的一次性語(yǔ)義。
• 2019 年更多的是 connector 對(duì)其它 API 的一些補(bǔ)完，包括對(duì)批的讀取以及 Table API 都有了支持。
• 2020 年的主要關(guān)注點(diǎn)是對(duì) Flink 1.11 的集成，其中的重點(diǎn)是 FLIP-27 以及 FLIP-95 的新特性集成。

2. Checkpoint 集成實(shí)現(xiàn)

以 Kafka 為例，可以首先來(lái)看一下 Kafka 是如何做到 Flink Checkpoint 的集成的。

上圖所示是一個(gè)典型的 Kafka "讀" 的架構(gòu)。基于 Chandy-Lamport 算法的 Flink checkpoint 實(shí)現(xiàn)，當(dāng) Job master Trigger 一個(gè) Checkpoint 時(shí)，會(huì)往 Task Executor 發(fā)送 RPC 請(qǐng)求。其接收到之后會(huì)把自身狀態(tài)存儲(chǔ)中的 Kafka commit offset 合并回 Job Manager 形成一個(gè) Checkpoint Metadata。

仔細(xì)思考后，其實(shí)可以發(fā)現(xiàn)其中的一些小問(wèn)題：

• 擴(kuò)縮容以及動(dòng)態(tài)的平衡支持。當(dāng) Partition 進(jìn)行調(diào)整的時(shí)候，或者說(shuō)對(duì) Pravega 而言，在 Partition 動(dòng)態(tài)擴(kuò)容和縮容的時(shí)候，如何進(jìn)行 Merge 一致性的保證。
• 還有一點(diǎn)就是 Task 需要維護(hù)一個(gè) offset 的信息，整個(gè)設(shè)計(jì)會(huì)與 Kafka 的內(nèi)部抽象 offset 耦合。

基于這些不足之處，Pravega 有自己內(nèi)部設(shè)計(jì)的 Checkpoint 機(jī)制，我們來(lái)看一下它是怎么和 Flink 的 Checkpoint 進(jìn)行集成的。

同樣讀取 Pravega Stream。開(kāi)始 Checkpoint 這里就有不同，Job master 不再向 Task Executor 發(fā)送 RPC 請(qǐng)求，轉(zhuǎn)而以 ExternallyInducedSource 的接口，向 Pravega 發(fā)送一個(gè) Checkpoint 的請(qǐng)求。

同時(shí)，Pravega 內(nèi)部會(huì)利用 StateSynchronizer 組件來(lái)同步和協(xié)調(diào)所有的 reader，并且會(huì)在所有的 reader 之間，發(fā)送 Checkpoint 的 event。當(dāng) Task Executor 讀到 Checkpoint Event 之后，整個(gè) Pravega 會(huì)標(biāo)志著這個(gè) Checkpoint 完成，然后返回的 Pravega Checkpoint 會(huì)存到 Job master state 當(dāng)中，從而完成 Checkpoint。

這樣的實(shí)現(xiàn)其實(shí)對(duì)于 Flink 來(lái)說(shuō)是更干凈的，因?yàn)樗鼪](méi)有耦合外部系統(tǒng)的實(shí)現(xiàn)細(xì)節(jié)，整個(gè) Checkpoint 的工作是交給 Pravega 來(lái)實(shí)現(xiàn)并完成的。

三、回顧 Flink 1.11 高階特性心得分享

Flink1.11 是 2020 年的一個(gè)重要發(fā)布版本，對(duì) connector 而言其實(shí)也有非常多的挑戰(zhàn)，主要集中在兩個(gè) FLIP 的實(shí)現(xiàn)：FLIP-27 以及 FLIP-95。對(duì)于這兩個(gè)全新功能，團(tuán)隊(duì)也花了很多時(shí)間去集成，在過(guò)程中也遇到了一些問(wèn)題和挑戰(zhàn)。下面我們來(lái)向大家分享一下我們是如何踩坑和填坑的。本文會(huì)以 FLIP-95 為例展開(kāi)。

1. FLIP-95 集成

FLIP-95 是新的 Table API，其動(dòng)機(jī)和 FLIP-27 類似，也是為了實(shí)現(xiàn)批流一體的接口，同時(shí)也能更好地支持 CDC 的集成。針對(duì)冗長(zhǎng)的配置鍵，也提出了相應(yīng)的 FLIP-122 來(lái)簡(jiǎn)化配置鍵的設(shè)定。

• 1.1 Pravega 舊的 Table API

從上圖可以看到 Pravega 在 Flink 1.10 之前的一個(gè) Table API，并且從圖中建表的 DDL 可以看到：

• 以 update mode 和 append 去進(jìn)行區(qū)分批和流，而且批流的數(shù)據(jù)這樣的區(qū)分并不直觀。
• 配置件也非常的冗長(zhǎng)和復(fù)雜，讀取的 Stream 需要通過(guò) connector.reader.stream-info.0 這樣非常長(zhǎng)的配置鍵來(lái)配置。
• 在代碼層面，和 DataStream API 也有非常多的耦合難以維護(hù)。

針對(duì)這些問(wèn)題，我們也就有了非常大的動(dòng)力去實(shí)現(xiàn)這樣一套新的 API，讓用戶更好的去使用表的抽象。整個(gè)框架如圖所示，借由整個(gè)新框架的幫助，所有的配置項(xiàng)通過(guò) ConfigOption 接口定義，并且都集中在 PravegaOptions 類管理。

• 1.2 Pravega 全新 Table API

下圖是最新 Table API 建表的實(shí)現(xiàn)，和之前的相比有非常大的簡(jiǎn)化，同時(shí)在功能上也有了不少優(yōu)化，例如企業(yè)級(jí)安全選項(xiàng)的配置，多 stream 以及起始 streamcut 的指定功能。

2. Flink-18641 解決過(guò)程心得分享

接下來(lái)，我想在此分享 Flink 1.11 集成的一個(gè)小的心得，是關(guān)于一個(gè) issue 解決過(guò)程的分享。Flink-18641 是我們?cè)诩?1.11.0 版本時(shí)碰到的問(wèn)題。升級(jí)的過(guò)程中，在單元測(cè)試中會(huì)報(bào) CheckpointException。接下來(lái)是我們完整的 debug 過(guò)程。

• 首先會(huì)自己去逐步斷點(diǎn)調(diào)試，通過(guò)查看 error 的報(bào)錯(cuò)日志，分析相關(guān)的 Pravega 以及 Flink 的源碼，確定它是 Flink CheckpointCoordinator 相關(guān)的一些問(wèn)題;
• 然后我們也查看了社區(qū)的一些提交記錄，發(fā)現(xiàn) Flink 1.10 之后， CheckpointCoordinator 線程模型，由原來(lái)鎖控制的模型變成了 Mailbox 模型。這個(gè)模型導(dǎo)致了我們?cè)瓉?lái)同步串型化執(zhí)行的一些邏輯，錯(cuò)誤的被并行化運(yùn)行了，于是導(dǎo)致該錯(cuò)誤;
• 進(jìn)一步看了這一個(gè)改動(dòng)的 pull request，也通過(guò)郵件和相關(guān)的一些 Committer 取得了聯(lián)系。最后在 dev 郵件列表上確認(rèn)問(wèn)題，并且開(kāi)了這個(gè) JIRA ticket。

我們也總結(jié)了以下一些注意事項(xiàng)給到在做開(kāi)源社區(qū)的同胞們：

• 在郵件列表和 JIRA 中搜索是否有其他人已經(jīng)提出了類似問(wèn)題;
• 完整的描述問(wèn)題，提供詳細(xì)的版本信息，報(bào)錯(cuò)日志和重現(xiàn)步驟;
• 得到社區(qū)成員反饋之后，可以進(jìn)一步會(huì)議溝通商討解決方案;
• 在非中文環(huán)境需要使用英語(yǔ)。

其實(shí)作為中國(guó)的開(kāi)發(fā)人員，有除了像 mailing list 和 JIRA 之外。我們也有釘釘群以及視頻的方式可以聯(lián)系到非常多的 Committer。其實(shí)更多的就是一個(gè)交流的過(guò)程，做開(kāi)源就是要和社區(qū)多交流，可以促進(jìn)項(xiàng)目之間的共同成長(zhǎng)。

四、未來(lái)展望

• 在未來(lái)比較大的工作就是 Pravega schema registry 集成。Pravega schema registry 提供了對(duì) Pravega stream 的元數(shù)據(jù)的管理，包括數(shù)據(jù) schema 以及序列化方式，并進(jìn)行存儲(chǔ)。這個(gè)功能伴隨著 Pravega 0.8 版本發(fā)布了該項(xiàng)目的第一個(gè)開(kāi)源版本。我們將在之后的 0.10 版本中基于這一項(xiàng)目實(shí)現(xiàn) Pravega 的Catalog，使得 Flink table API 的使用更加簡(jiǎn)單;
• 其次，我們也時(shí)刻關(guān)注 Flink 社區(qū)的新動(dòng)向，對(duì)于社區(qū)的新版本、新功能也會(huì)積極集成，目前的計(jì)劃包括 FLIP-143 和 FLIP-129;
• 社區(qū)也在逐步完成基于 docker 容器的新的 Test Framework 的轉(zhuǎn)換，我們也在關(guān)注并進(jìn)行集成。

最后也希望社區(qū)的小伙伴可以多多的關(guān)注 Pravega 項(xiàng)目，促進(jìn) Pravega connector 與 Flink 的共同發(fā)展。