Netflix 是如何實(shí)現(xiàn)每秒 200 萬次的數(shù)據(jù)處理，并查詢超過 1.5 萬億行的數(shù)據(jù)？

在推動(dòng)技術(shù)創(chuàng)新升級(jí)的同時(shí)，還要確保 Netflix 始終如一的良好體驗(yàn)，這并非易事。

如何才能確保更新不會(huì)影響到用戶呢？如果確保我們的改進(jìn)是可度量的呢？Netflix 使用來自回放設(shè)備的實(shí)時(shí)日志作為事件源來獲得度量，以便理解和量化用戶設(shè)備瀏覽和回放的流暢度。

一旦有了這些度量，我們就把它們輸入數(shù)據(jù)庫。每一項(xiàng)指標(biāo)都附有與所使用設(shè)備類型相關(guān)的匿名細(xì)節(jié)，例如，該設(shè)備是智能電視、iPad 還是 Android 手機(jī)。這樣，我們就可以對(duì)設(shè)備進(jìn)行分類，并從不同的方面來查看數(shù)據(jù)。同樣，我們還能夠只隔離影響特定群體的問題，如應(yīng)用的版本、特定類型的設(shè)備或特定國(guó)家。

這些聚合數(shù)據(jù)可以立即用于查詢，可以通過儀表板查詢，也可以通過即席查詢。這些指標(biāo)還會(huì)持續(xù)檢查報(bào)警信號(hào)，比如新版本是否會(huì)影響某些用戶或設(shè)備的回放或?yàn)g覽。這些檢查用于通知負(fù)責(zé)的團(tuán)隊(duì)，讓他們可以盡快處理問題。

在軟件更新期間，我們?yōu)橐徊糠钟脩魡⒂眯掳姹?，并使用這些實(shí)時(shí)指標(biāo)來比較新版本與舊版本的性能。在度量中，如果有任何不合適，我們可以中止更新并將那些已獲得新版本的用戶恢復(fù)到以前的版本。

由于這些數(shù)據(jù)的處理速度超過每秒 200 萬次，所以將其存入一個(gè)可以快速查詢的數(shù)據(jù)庫非常困難。我們需要足夠的數(shù)據(jù)維數(shù)，以便能夠有效地隔離問題，如此一來，我們每天生成超過 1150 億行數(shù)據(jù)。在 Netflix，我們利用 Apache Druid 幫助我們?cè)谶@種規(guī)模下解決這一挑戰(zhàn)。

1. Druid

Apache Druid 是一個(gè)高性能的實(shí)時(shí)分析數(shù)據(jù)庫。它是針對(duì)特別注重快速查詢和攝取的工作流而設(shè)計(jì)。Druid 特別適合于即時(shí)的數(shù)據(jù)可視化、即席查詢、操作分析和高并發(fā)處理。——druid.io

因此，Druid 非常適合我們的用例，事件數(shù)據(jù)攝取率很高，而且具有高基數(shù)（high cardinality）和快速查詢需求。

Druid 不是一個(gè)關(guān)系型數(shù)據(jù)庫，但是一些概念是可以轉(zhuǎn)化的。我們有數(shù)據(jù)源，而不是表。與關(guān)系型數(shù)據(jù)庫一樣，有表示為列的數(shù)據(jù)邏輯分組。與關(guān)系型數(shù)據(jù)庫不同的是，沒有連接的概念。因此，我們需要確保在每個(gè)數(shù)據(jù)源中都包含希望的篩選或分組的列。

數(shù)據(jù)源中主要有三種列——時(shí)間、維度和度量。

Druid 中的一切都有時(shí)間標(biāo)記。每個(gè)數(shù)據(jù)源都有一個(gè)時(shí)間戳列，這是主要的分區(qū)機(jī)制。維度是可用于篩選、查詢或分組的值。度量是可以聚合的值，并且?guī)缀蹩偸菙?shù)值。

通過移除執(zhí)行連接的能力，并假設(shè)數(shù)據(jù)都有時(shí)間戳，Druid 可以在存儲(chǔ)、分發(fā)和查詢數(shù)據(jù)方面做一些優(yōu)化，這樣我們就可以將數(shù)據(jù)源擴(kuò)展到數(shù)萬億行，并且仍然可以實(shí)現(xiàn)查詢響應(yīng)時(shí)間在 10 毫秒以內(nèi)。

為了達(dá)到這種程度的可擴(kuò)展性，Druid 把存儲(chǔ)的數(shù)據(jù)分成時(shí)間塊。時(shí)間塊的長(zhǎng)度是可配置的?？梢愿鶕?jù)數(shù)據(jù)和用例選擇適當(dāng)?shù)膮^(qū)間。對(duì)于數(shù)據(jù)和用例，我們使用 1 小時(shí)的時(shí)間塊。時(shí)間塊中的數(shù)據(jù)存儲(chǔ)在一個(gè)或多個(gè) 段 中。每個(gè)段包含所有屬于這個(gè)時(shí)間塊的數(shù)據(jù)行，時(shí)間塊由它的時(shí)間戳列決定。段的大小可以配置為行數(shù)上限或段文件的總大小。

在查詢數(shù)據(jù)時(shí)，Druid 將查詢發(fā)送到集群中所有那些擁有的段所屬的時(shí)間塊在查詢范圍內(nèi)的節(jié)點(diǎn)。在將中間結(jié)果發(fā)送回查詢代理節(jié)點(diǎn)之前，每個(gè)節(jié)點(diǎn)都并行地針對(duì)其持有的數(shù)據(jù)處理查詢。在將結(jié)果集發(fā)送回客戶端之前，代理將執(zhí)行最后的合并和聚合。

2. 攝取

這個(gè)數(shù)據(jù)庫的數(shù)據(jù)插入是實(shí)時(shí)的，不是將單個(gè)記錄插入到數(shù)據(jù)源中，而是從 Kafka 流讀取事件（就是我們的度量）。每個(gè)數(shù)據(jù)源使用一個(gè)主題。在 Druid 中，我們使用 Kafka 索引任務(wù)，它創(chuàng)建了多個(gè)分布在實(shí)時(shí)節(jié)點(diǎn)（中間管理器）上的索引工作器。

這些索引器都訂閱主題，并從流中讀取其事件。索引器根據(jù)攝取規(guī)范從事件消息中提取值，并將創(chuàng)建的行累積到內(nèi)存中。一旦創(chuàng)建了一行，就可以查詢它。對(duì)于索引器正在填充的段的時(shí)間塊進(jìn)行查詢，將由索引器本身提供服務(wù)。由于索引任務(wù)本質(zhì)上是執(zhí)行兩項(xiàng)工作，即攝取和處理查詢，所以及時(shí)將數(shù)據(jù)發(fā)送到歷史節(jié)點(diǎn)，以更優(yōu)化的方式將查詢工作卸載給它們是很重要的。

Druid 可以在攝取時(shí)匯總數(shù)據(jù)，以盡量減少需要存儲(chǔ)的原始數(shù)據(jù)量。Rollup 是一種匯總或預(yù)聚合的形式。在某些情況下，匯總數(shù)據(jù)可以極大地減少需要存儲(chǔ)的數(shù)據(jù)的大小，可能會(huì)減少行數(shù)數(shù)量級(jí)。然而，這種存儲(chǔ)減少是有代價(jià)的：我們失去了查詢單個(gè)事件的能力，只能在預(yù)定義的查詢粒度上進(jìn)行查詢。對(duì)于我們的用例，我們選擇了 1 分鐘的查詢粒度。

在攝取期間，如果任何行具有相同的維度，并且它們的時(shí)間戳在同一分鐘內(nèi)（我們的查詢粒度），則將這些行匯總。這意味著，通過將所有度量值相加合并行并增加計(jì)數(shù)器，我們就可以知道有多少事件對(duì)這一行的值有貢獻(xiàn)。這種形式的 Rollup 可以顯著地減少數(shù)據(jù)庫中的行數(shù)，從而加快查詢速度。

一旦累積的行數(shù)達(dá)到某個(gè)閾值，或者段打開的時(shí)間太長(zhǎng)，這些行就被寫入段文件并被卸載到深層存儲(chǔ)中。然后，索引器通知協(xié)調(diào)器片段已經(jīng)做好準(zhǔn)備，以便協(xié)調(diào)器可以告訴一個(gè)或多個(gè)歷史節(jié)點(diǎn)來加載它。一旦段被成功地加載到歷史節(jié)點(diǎn)中，它就會(huì)從索引器中卸載，任何針對(duì)該數(shù)據(jù)的查詢現(xiàn)在都將由歷史節(jié)點(diǎn)提供服務(wù)。

3. 數(shù)據(jù)管理

可以想象，隨著維度基數(shù)的增加，在同一分鐘內(nèi)發(fā)生相同事件的可能性會(huì)降低。管理基數(shù)（以便匯總）是實(shí)現(xiàn)良好查詢性能的強(qiáng)大手段。

為了達(dá)到我們需要的攝取速度，可以運(yùn)行許多索引器實(shí)例。即使索引任務(wù)使用 Rollup 合并相同的行，在一個(gè)索引任務(wù)的同一個(gè)實(shí)例中獲得這些相同行的機(jī)會(huì)也非常低。為了解決這個(gè)問題并實(shí)現(xiàn)盡可能好的 Rollup，我們會(huì)在給定時(shí)間塊的所有段都傳遞給歷史節(jié)點(diǎn)之后運(yùn)行一個(gè)任務(wù)。

預(yù)定的壓縮任務(wù)從深度存儲(chǔ)中獲取時(shí)間塊的所有段，并運(yùn)行 map/reduce 作業(yè)來重新創(chuàng)建段并實(shí)現(xiàn)完美的匯總。然后，由歷史節(jié)點(diǎn)加載和發(fā)布新的段，替換和取代原來的、未充分匯總的段。在我們的例子中，通過使用這個(gè)額外的壓縮任務(wù)，行數(shù)減少到了 1/2。

知道何時(shí)收到給定時(shí)間塊的所有事件并不是一件小事。Kafka 上可能有延遲到達(dá)的數(shù)據(jù)，或者索引器將片段傳遞給歷史節(jié)點(diǎn)可能需要花些時(shí)間。為了解決這個(gè)問題，我們會(huì)在運(yùn)行壓縮之前執(zhí)行一些限制和檢查。

首先，我們丟棄所有非常晚才到達(dá)的數(shù)據(jù)。我們認(rèn)為，這些數(shù)據(jù)在我們的實(shí)時(shí)系統(tǒng)已經(jīng)過時(shí)。這設(shè)置了數(shù)據(jù)延遲的界限。其次，壓縮任務(wù)被延遲調(diào)度，這使得段有足夠的時(shí)間可以卸載到正常流中的歷史節(jié)點(diǎn)。最后，當(dāng)給定時(shí)間塊的預(yù)定壓縮任務(wù)啟動(dòng)時(shí)，它將查詢段元數(shù)據(jù)，檢查是否仍然有相關(guān)的段被寫入或傳遞。如果有，它將等待幾分鐘后再試一次。這將確保所有數(shù)據(jù)都由壓縮作業(yè)處理。

沒有這些措施，我們發(fā)現(xiàn)有時(shí)會(huì)丟失數(shù)據(jù)。在開始?jí)嚎s時(shí)仍有寫入的段將被新壓縮的段所覆蓋，這些段具有更高的版本，因此會(huì)優(yōu)先。這可以有效地刪除包含在那些尚未完成傳遞的段中的數(shù)據(jù)。

4. 查詢

Druid 支持兩種查詢語言：Druid SQL 和原生查詢。在底層，Druid SQL 查詢會(huì)被轉(zhuǎn)換成原生查詢。原生查詢以 JSON 格式提交給 REST 端點(diǎn)，這是我們使用的主要機(jī)制。

我們集群的大多數(shù)查詢都是由自定義的內(nèi)部工具（如儀表板和預(yù)警系統(tǒng)）生成的。這些系統(tǒng)最初是為了與我們內(nèi)部開發(fā)的開源時(shí)序數(shù)據(jù)庫 Atlas 一起工作而設(shè)計(jì)的。因此，這些工具使用 Atlas Stack 查詢語言。

為了加速查詢 Druid 的采用，并實(shí)現(xiàn)現(xiàn)有工具的重用，我們添加了一個(gè)翻譯層來接收 Atlas 查詢，將它們重寫為 Druid 查詢，發(fā)送查詢并將結(jié)果重新格式化為 Atlas 結(jié)果。這個(gè)抽象層允許現(xiàn)有的工具按原樣使用，用戶要訪問我們 Druid 數(shù)據(jù)存儲(chǔ)中的數(shù)據(jù)也不需要額外學(xué)習(xí)。

5. 調(diào)優(yōu)

在調(diào)整集群節(jié)點(diǎn)的配置時(shí)，我們以較高的速度運(yùn)行一系列可重復(fù)和可預(yù)測(cè)的查詢，從而獲得每個(gè)給定配置的響應(yīng)時(shí)間和查詢吞吐量的基準(zhǔn)。這些查詢?cè)谠O(shè)計(jì)時(shí)隔離了集群的各個(gè)部分，以檢查查詢性能方面的改善或退化。

例如，我們對(duì)最近的數(shù)據(jù)進(jìn)行有針對(duì)性的查詢，以便只對(duì) Middle Manager 進(jìn)行查詢。同樣，對(duì)于較長(zhǎng)的時(shí)間段但較舊的數(shù)據(jù)，我們只查詢歷史節(jié)點(diǎn)來測(cè)試緩存配置。同樣，使用按高基數(shù)維分組的查詢檢查結(jié)果合并受到了什么影響。我們繼續(xù)調(diào)整和運(yùn)行這些基準(zhǔn)測(cè)試，直到我們對(duì)查詢性能滿意為止。

在這些測(cè)試中，我們發(fā)現(xiàn)調(diào)整緩沖區(qū)的大小、線程的數(shù)量、查詢隊(duì)列的長(zhǎng)度和分配給查詢緩存的內(nèi)存對(duì)查詢性能有實(shí)際的影響。然而，壓縮作業(yè)的引入對(duì)查詢性能有更重要的影響，它會(huì)將未充分匯總的段重新壓縮，實(shí)現(xiàn)完美匯總。

我們還發(fā)現(xiàn)，在歷史節(jié)點(diǎn)上啟用緩存非常有好處，而在代理節(jié)點(diǎn)上啟用緩存效果則不是很明顯。因此，我們不在代理上使用緩存。這可能是由我們的用例造成的，但是幾乎每一次查詢都會(huì)錯(cuò)過代理上的緩存，這可能是因?yàn)椴樵兺ǔ０钚碌臄?shù)據(jù)，這些數(shù)據(jù)不在任何緩存中，因?yàn)橐恢庇袛?shù)據(jù)到達(dá)。

6. 小結(jié)

針對(duì)我們的用例和數(shù)據(jù)率，經(jīng)過多次優(yōu)化調(diào)整，Druid 已經(jīng)被證明具備我們最初希望的能力。

我們已經(jīng)能夠得到一個(gè)能力齊備、可用的系統(tǒng)，但仍然有更多的工作要做。隨著查詢數(shù)量和復(fù)雜性的增加，我們的攝取量和速率也在不斷增加。隨著更多的團(tuán)隊(duì)認(rèn)識(shí)到這些詳細(xì)數(shù)據(jù)的價(jià)值，我們經(jīng)常需要添加更多的度量和維度，這加重了系統(tǒng)的負(fù)擔(dān)。我們必須繼續(xù)監(jiān)控和調(diào)優(yōu)，保證查詢性能受控。

目前，我們正在以每秒 200 萬次的速度處理事件，并查詢超過 1.5 萬億行的數(shù)據(jù)，以獲得關(guān)于用戶體驗(yàn)服務(wù)的詳細(xì)信息。這一切幫助我們保持了高質(zhì)量的 Netflix 體驗(yàn)，同時(shí)能夠不斷地創(chuàng)新。