如何選擇適合自己業(yè)務(wù)的流計(jì)算引擎?除了比較各自的功能矩陣外，基準(zhǔn)測試(benchmark)便是用來評估系統(tǒng)性能的一個重要和常見的方法。然而在流計(jì)算領(lǐng)域，目前還沒有一個行業(yè)標(biāo)準(zhǔn)的基準(zhǔn)測試。本文將探討流計(jì)算基準(zhǔn)測試設(shè)計(jì)上的難點(diǎn)，分享如何設(shè)計(jì)流計(jì)算基準(zhǔn)測試框架——Nexmark，以及將來的規(guī)劃。

一 背景

隨著數(shù)據(jù)時效性對企業(yè)的精細(xì)化運(yùn)營越來越重要，“實(shí)時即未來”、“實(shí)時數(shù)倉”、“數(shù)據(jù)湖” 成為了近幾年炙手可熱的詞。流計(jì)算領(lǐng)域的格局也在這幾年發(fā)生了巨大的變化，Apache Flink 在流批一體的方向上不斷深耕，Apache Spark 的近實(shí)時處理有著一定的受眾，Apache Kafka 也有了 ksqlDB 高調(diào)地進(jìn)軍流計(jì)算，而 Apache Storm 卻開始逐漸地退出歷史的舞臺。

每一種引擎有其優(yōu)勢的地方，如何選擇適合自己業(yè)務(wù)的流計(jì)算引擎成了一個由來已久的話題。除了比較各個引擎提供的不同的功能矩陣之外，性能是一個無法繞開的評估因素?；鶞?zhǔn)測試(benchmark)就是用來評估系統(tǒng)性能的一個重要和常見的過程。

二 現(xiàn)有流計(jì)算基準(zhǔn)測試的問題

目前在流計(jì)算領(lǐng)域中，還沒有一個行業(yè)標(biāo)準(zhǔn)的基準(zhǔn)測試。目前業(yè)界較為人知的流計(jì)算 benchmark 是五年前雅虎 Storm 團(tuán)隊(duì)發(fā)布的 Yahoo Streaming Benchmarks[4]。雅虎的原意是因?yàn)闃I(yè)界缺少反映真實(shí)場景的 benchmark，模擬了一個簡單的廣告場景來比較各個流計(jì)算框架，后來被廣泛引用。具體場景是從 Kafka 消費(fèi)的廣告的點(diǎn)擊流，關(guān)聯(lián) Redis 中的廣告所屬的 campaign 信息，然后做時間窗口聚合計(jì)數(shù)。

然而，正是因?yàn)檠呕F(tuán)隊(duì)太過于追求還原真實(shí)的生產(chǎn)環(huán)境，導(dǎo)致這些外部系統(tǒng)服務(wù)(Kafka, Redis)成為了作業(yè)的瓶頸。Ververica 曾在這篇文章[5]中做過一個擴(kuò)展實(shí)驗(yàn)，將數(shù)據(jù)源從 Kafka 替換成了一個內(nèi)置的 datagen source，性能提升了 37 倍!由此可見，引入的 Kafka 組件導(dǎo)致了無法準(zhǔn)確反映引擎真實(shí)的性能。更重要的一個問題是，Yahoo Benchmark 只包含一個非常簡單的，類似 “Word Count” 的作業(yè)，它無法全面地反映當(dāng)今復(fù)雜的流計(jì)算系統(tǒng)和業(yè)務(wù)。試想，誰會用一個簡單的 “Word Count” 去衡量比較各個數(shù)據(jù)庫之間的性能差異呢?正是這些原因使得 Yahoo Benchmark 無法成為一個行業(yè)標(biāo)準(zhǔn)的基準(zhǔn)測試。這也正是我們想要解決的問題。

因此，我們認(rèn)為一個行業(yè)標(biāo)準(zhǔn)的基準(zhǔn)測試應(yīng)該具備以下幾個特點(diǎn)：

可復(fù)現(xiàn)性

可復(fù)現(xiàn)性是使得 benchmark 被信任的一個重要條件。許多 benchmark 的結(jié)果是難以重現(xiàn)的。有的是因?yàn)橹粩[了個 benchmark 結(jié)果圖，用于生成這些結(jié)果的代碼并沒有公開。有的是因?yàn)橛糜?benchmark 的硬件不容易被別人獲取到。有的是因?yàn)?benchmark 依賴的服務(wù)太多，致使測試結(jié)果不穩(wěn)定。

能代表和覆蓋行業(yè)真實(shí)的業(yè)務(wù)場景( query 量)

例如數(shù)據(jù)庫領(lǐng)域非常著名的 TPC-H、TPC-DS 涵蓋了大量的 query 集合，來捕獲查詢引擎之間細(xì)微的差別。而且這些 query 集合都立于真實(shí)業(yè)務(wù)場景之上(商品零售行業(yè))，數(shù)據(jù)規(guī)模大，因此也很受一些大數(shù)據(jù)系統(tǒng)的青睞。

能調(diào)整作業(yè)的負(fù)載(數(shù)據(jù)量、數(shù)據(jù)分布)

在大數(shù)據(jù)領(lǐng)域，不同的數(shù)據(jù)規(guī)模對于引擎來說可能會是完全不同的事情。例如 Yahoo Benchmark 中使用的 campaign id 只有 100 個，使得狀態(tài)非常小，內(nèi)存都可以裝的下。這樣使得同步 IO 和 checkpoint 等的影響可以忽略不計(jì)。而真實(shí)的場景往往要面對大狀態(tài)，面臨的挑戰(zhàn)要復(fù)雜困難的多。像 TPC-DS 的數(shù)據(jù)生成工具會提供 scalar factor 的參數(shù)來控制數(shù)據(jù)量。其次在數(shù)據(jù)分布上最好也能貼近真實(shí)世界的數(shù)據(jù)，如有數(shù)據(jù)傾斜，及調(diào)整傾斜比例。從而能全面、綜合地反映業(yè)務(wù)場景和引擎之間地差異。

有統(tǒng)一的性能衡量指標(biāo)和采集匯總工具

基準(zhǔn)測試的性能指標(biāo)的定義需要清晰、一致，且能適用于各種計(jì)算引擎。然而流計(jì)算的性能指標(biāo)要比傳統(tǒng)批處理的更難定義、更難采集。是流計(jì)算 benchmark 最具挑戰(zhàn)性的一個問題，這也會在下文展開描述。

我們也研究了很多其他的流計(jì)算相關(guān)的基準(zhǔn)測試，包括：StreamBench、HiBench、BigDataBench，但是它們都在上述幾個基本面有所欠缺?；鶞?zhǔn)測試的行業(yè)標(biāo)桿無疑是 TPC 發(fā)布的一系列 benchmark，如 TPC-H，TPC-DS。然而這些 benchmark 是面向傳統(tǒng)數(shù)據(jù)庫、傳統(tǒng)數(shù)倉而設(shè)計(jì)的，并不適用于今天的流計(jì)算系統(tǒng)。例如 benchmark 中沒有考慮事件時間、數(shù)據(jù)的亂序、窗口等流計(jì)算中常見的場景。因此我們不得不考慮重新設(shè)計(jì)并開源一個流計(jì)算基準(zhǔn)測試框架——Nexmark。

地址：https://github.com/nexmark/nexmark。

三 Nexmark 基準(zhǔn)測試框架的設(shè)計(jì)

為了提供一個滿足以上幾個基本面的流計(jì)算基準(zhǔn)測試，我們設(shè)計(jì)和開發(fā)了 Nexmark 基準(zhǔn)測試框架，并努力讓其成為流計(jì)算領(lǐng)域的標(biāo)準(zhǔn) benchmark 。

Nexmark 基準(zhǔn)測試框架來源于 NEXMark 研究論文[1]，以及 Apache Beam Nexmark Suite[6]，并在其之上進(jìn)行了擴(kuò)展和完善。Nexmark 基準(zhǔn)測試框架不依賴任何第三方服務(wù)，只需要部署好引擎和 Nexmark，通過腳本 nexmark/bin/run_query.sh all 即可等待并獲得所有 query 下的 benchmark 結(jié)果。下面我們將探討 Nexmark 基準(zhǔn)測試在設(shè)計(jì)上的一些決策。

1 移除外部 source、sink 依賴

如上所述，Yahoo Benchmark 使用了 Kafka 數(shù)據(jù)源，卻使得最終結(jié)果無法準(zhǔn)確反映引擎的真實(shí)性能。此外，我們還發(fā)現(xiàn)，在 benchmark 快慢流雙流 JOIN 的場景時，如果使用了 Kafka 數(shù)據(jù)源，慢流會超前消費(fèi)(快流易被反壓)，導(dǎo)致 JOIN 節(jié)點(diǎn)的狀態(tài)會緩存大量超前的數(shù)據(jù)。這其實(shí)不能反映真實(shí)的場景，因?yàn)樵谡鎸?shí)的場景下，慢流是無法被超前消費(fèi)的(數(shù)據(jù)還未產(chǎn)生)。所以我們在 Nexmark 中使用了 datagen source，數(shù)據(jù)直接在內(nèi)存中生成，數(shù)據(jù)不落地，直接向下游節(jié)點(diǎn)發(fā)送。多個事件流都由單一的數(shù)據(jù)生成器生成，所以當(dāng)快流被反壓時，也能抑制慢流的生成，較好地反映了真實(shí)場景。

與之類似的，我們也移除了外部 sink 的依賴，不再輸出到 Kafka/Redis，而是輸出到一個空 sink 中，即 sink 會丟棄收到的所有數(shù)據(jù)。

通過這種方式，我們保證了瓶頸只會在引擎自身，從而能精確地測量出引擎之間細(xì)微的差異。

2 Metrics

批處理系統(tǒng) benchmark 的 metric 通常采用總體耗時來衡量。然而流計(jì)算系統(tǒng)處理的數(shù)據(jù)是源源不斷的，無法統(tǒng)計(jì) query 耗時。因此，我們提出三個主要的 metric：吞吐、延遲、CPU。Nexmark 測試框架會自動幫我們采集 metric，并做匯總，不需要部署任何第三方的 metric 服務(wù)。

吞吐

吞吐(throughput)也常被稱作 TPS，描述流計(jì)算系統(tǒng)每秒能處理多少條數(shù)據(jù)。由于我們有多個事件流，所有事件流都由一個數(shù)據(jù)生成器生成，為了統(tǒng)一觀測角度，我們采用數(shù)據(jù)生成器的 TPS，而非單一事件流的 TPS。我們將一個 query 能達(dá)到的最大吞吐，作為其吞吐指標(biāo)。例如，針對 Flink 引擎，我們通過 Flink REST API 暴露的.numRecordsOutPerSecond metric 來獲取當(dāng)前吞吐量。

延遲

延遲(Latency)描述了從數(shù)據(jù)進(jìn)入流計(jì)算系統(tǒng)，到它的結(jié)果被輸出的時間間隔。對于窗口聚合，Yahoo Benchmark 中使用 output_system_time - window_end 作為延遲指標(biāo)，這其實(shí)并沒有考慮數(shù)據(jù)在窗口輸出前的等待時間，這種計(jì)算結(jié)果也會極大地受到反壓的影響，所以其計(jì)算結(jié)果是不準(zhǔn)確的。一種更準(zhǔn)確的計(jì)算方式應(yīng)為 output_system_time - max(ingest_time)。然而在非窗口聚合，或雙流 JOIN 中，延遲又會有不同的計(jì)算方式。

所以延遲的定義和采集在流計(jì)算系統(tǒng)中有很多現(xiàn)實(shí)存在的問題，需要根據(jù)具體 query 具體分析，這在參考文獻(xiàn)[2]中有詳細(xì)的討論，這也是我們目前還未在 Nexmark 中實(shí)現(xiàn)延遲 metric 的原因。

CPU

資源使用率是很多流計(jì)算 benchmark 中忽視的一個指標(biāo)。由于在真實(shí)生產(chǎn)環(huán)境，我們并不會限制流計(jì)算引擎所能使用的核數(shù)，從而給系統(tǒng)更大的彈性。所以我們引入了 CPU 使用率，作為輔助指標(biāo)，即作業(yè)一共消耗了多少核。通過吞吐/cores，可以計(jì)算出平均每個核對于吞吐的貢獻(xiàn)。對于進(jìn)程的 CPU 使用率的采集，我們沒有使用 JVM CPU load，而是借鑒了 YARN 中的實(shí)現(xiàn)，通過采樣/proc/ /stat 并計(jì)算獲得，該方式可以獲得較為真實(shí)的進(jìn)程 CPU 使用率。因此我們的 Nexmark 測試框架需要在測試開始前，先在每臺機(jī)器上部署 CPU 采集進(jìn)程。

3 Query 與 Schema

Nexmark 的業(yè)務(wù)模型基于一個真實(shí)的在線拍賣系統(tǒng)。所有的 query 都基于相同的三個數(shù)據(jù)流，三個數(shù)據(jù)流會有一個數(shù)據(jù)生成器生成，來控制他們之間的比例、數(shù)據(jù)偏斜、關(guān)聯(lián)關(guān)系等等。這三個數(shù)據(jù)流分別是：

• 用戶(Person)：代表一個提交拍賣，或參與競標(biāo)的用戶。
• 拍賣(Auction)：代表一個拍賣品。
• 競標(biāo)(Bid)：代表一個對拍賣品的出價(jià)。

我們一共定義了 16 個 query，所有的 query 都使用 ANSI SQL 標(biāo)準(zhǔn)語法?；?SQL ，我們可以更容易地?cái)U(kuò)展 query 測試集，支持更多的引擎。然而，由于 Spark 在流計(jì)算功能上的限制，大部分的 query 都無法通過 Structured Streaming 來實(shí)現(xiàn)。因此我們目前只支持測試 Flink SQL 引擎。

??

4 作業(yè)負(fù)載的配置化

我們也支持配置調(diào)整作業(yè)的負(fù)載，包括數(shù)據(jù)生成器的吞吐量以及吞吐曲線、各個數(shù)據(jù)流之間的數(shù)據(jù)量比例、每個數(shù)據(jù)流的數(shù)據(jù)平均大小以及數(shù)據(jù)傾斜比例等等。具體的可以參考 Source DDL 參數(shù)。

四 實(shí)驗(yàn)結(jié)果

我們在阿里云的三臺機(jī)器上進(jìn)行了 Nexmark 針對 Flink 的基準(zhǔn)測試。每臺機(jī)器均為 ecs.i2g.2xlarge 規(guī)格，配有 Xeon 2.5 GHz CPU (8 vCores) 以及 32 GB 內(nèi)存，800 GB SSD 本地磁盤。機(jī)器之間的帶寬為 2 Gbps。

測試了 flink-1.11 版本，我們在這 3 臺機(jī)器上部署了 Flink standalone 集群，由 1 個 JobManager，8 個 TaskManager (每個只有 1 slot)組成，都是 4 GB內(nèi)存。集群默認(rèn)并行度為 8。開啟 checkpoint 以及 exactly once 模式，checkpoint 間隔 3 分鐘。使用 RocksDB 狀態(tài)后端。測試發(fā)現(xiàn)，對于有狀態(tài)的 query，每次 checkpoint 的大小在 GB 級以上，所以有效地測試的大狀態(tài)的場景。

Datagen source 保持 1000 萬每秒的速率生成數(shù)據(jù)，三個數(shù)據(jù)流的數(shù)據(jù)比例分別是 Bid: 92%，Auction: 6%，Person: 2%。每個 query 都先運(yùn)行 3 分鐘熱身，之后 3 分鐘采集性能指標(biāo)。

運(yùn)行 nexmark/bin/run_query.sh all 后，打印測試結(jié)果如下：

??

五 總結(jié)

我們開發(fā)和設(shè)計(jì) Nexmark 的初衷是為了推出一套標(biāo)準(zhǔn)的流計(jì)算 benchmark 測試集，以及測試流程。雖然目前僅支持了 Flink 引擎，但在當(dāng)前也具有一定的意義，例如：

推動流計(jì)算 benchmark 的發(fā)展和標(biāo)準(zhǔn)化。

作為 Flink 引擎版本迭代之間的性能測試工具，甚至是日常回歸工具，及時發(fā)現(xiàn)性能回退的問題。

在開發(fā) Flink 性能優(yōu)化的功能時，可以用來驗(yàn)證性能優(yōu)化的效果。

部分公司可能會有 Flink 的內(nèi)部版本，可以用作內(nèi)部版本與開源版本之間的性能對比工具。

當(dāng)然，我們也計(jì)劃持續(xù)改進(jìn)和完善 Nexmark 測試框架，例如支持 Latency metric，支持更多的引擎，如 Spark Structured Streaming, Spark Streaming, ksqlDB, Flink DataStream 等等。也歡迎有志之士一起加入貢獻(xiàn)和擴(kuò)展。

參考及引用

[1]Pete Tucker and Kristin Tufte. "NEXMark – A Benchmark for Queries over Data Streams". June 2010.[2]Jeyhun Karimov and Tilmann Rabl. "Benchmarking Distributed Stream Data Processing Systems". arXiv:1802.08496v2 [cs.DB] Jun 2019[3]Yangjun Wang. "Stream Processing Systems Benchmark: StreamBench". May 2016.[4]https://github.com/yahoo/streaming-benchmarks[5]https://www.ververica.com/blog/extending-the-yahoo-streaming-benchmark[6]https://beam.apache.org/documentation/sdks/java/testing/nexmark/