Netflix 是如何實現每秒 200 萬次的數據處理，并查詢超過 1.5 萬億行的數據？

在推動技術創新升級的同時，還要確保 Netflix 始終如一的良好體驗，這并非易事。

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

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

這些聚合數據可以立即用于查詢，可以通過儀表板查詢，也可以通過即席查詢。這些指標還會持續檢查報警信號，比如新版本是否會影響某些用戶或設備的回放或瀏覽。這些檢查用于通知負責的團隊，讓他們可以盡快處理問題。

在軟件更新期間，我們為一部分用戶啟用新版本，并使用這些實時指標來比較新版本與舊版本的性能。在度量中，如果有任何不合適，我們可以中止更新并將那些已獲得新版本的用戶恢復到以前的版本。

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

1. Druid

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

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

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

數據源中主要有三種列——時間、維度和度量。

Druid 中的一切都有時間標記。每個數據源都有一個時間戳列，這是主要的分區機制。維度是可用于篩選、查詢或分組的值。度量是可以聚合的值，并且幾乎總是數值。

通過移除執行連接的能力，并假設數據都有時間戳，Druid 可以在存儲、分發和查詢數據方面做一些優化，這樣我們就可以將數據源擴展到數萬億行，并且仍然可以實現查詢響應時間在 10 毫秒以內。

為了達到這種程度的可擴展性，Druid 把存儲的數據分成時間塊。時間塊的長度是可配置的。可以根據數據和用例選擇適當的區間。對于數據和用例，我們使用 1 小時的時間塊。時間塊中的數據存儲在一個或多個 段 中。每個段包含所有屬于這個時間塊的數據行，時間塊由它的時間戳列決定。段的大小可以配置為行數上限或段文件的總大小。

在查詢數據時，Druid 將查詢發送到集群中所有那些擁有的段所屬的時間塊在查詢范圍內的節點。在將中間結果發送回查詢代理節點之前，每個節點都并行地針對其持有的數據處理查詢。在將結果集發送回客戶端之前，代理將執行最后的合并和聚合。

2. 攝取

這個數據庫的數據插入是實時的，不是將單個記錄插入到數據源中，而是從 Kafka 流讀取事件（就是我們的度量）。每個數據源使用一個主題。在 Druid 中，我們使用 Kafka 索引任務，它創建了多個分布在實時節點（中間管理器）上的索引工作器。

這些索引器都訂閱主題，并從流中讀取其事件。索引器根據攝取規范從事件消息中提取值，并將創建的行累積到內存中。一旦創建了一行，就可以查詢它。對于索引器正在填充的段的時間塊進行查詢，將由索引器本身提供服務。由于索引任務本質上是執行兩項工作，即攝取和處理查詢，所以及時將數據發送到歷史節點，以更優化的方式將查詢工作卸載給它們是很重要的。

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

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

一旦累積的行數達到某個閾值，或者段打開的時間太長，這些行就被寫入段文件并被卸載到深層存儲中。然后，索引器通知協調器片段已經做好準備，以便協調器可以告訴一個或多個歷史節點來加載它。一旦段被成功地加載到歷史節點中，它就會從索引器中卸載，任何針對該數據的查詢現在都將由歷史節點提供服務。

3. 數據管理

可以想象，隨著維度基數的增加，在同一分鐘內發生相同事件的可能性會降低。管理基數（以便匯總）是實現良好查詢性能的強大手段。

為了達到我們需要的攝取速度，可以運行許多索引器實例。即使索引任務使用 Rollup 合并相同的行，在一個索引任務的同一個實例中獲得這些相同行的機會也非常低。為了解決這個問題并實現盡可能好的 Rollup，我們會在給定時間塊的所有段都傳遞給歷史節點之后運行一個任務。

預定的壓縮任務從深度存儲中獲取時間塊的所有段，并運行 map/reduce 作業來重新創建段并實現完美的匯總。然后，由歷史節點加載和發布新的段，替換和取代原來的、未充分匯總的段。在我們的例子中，通過使用這個額外的壓縮任務，行數減少到了 1/2。

知道何時收到給定時間塊的所有事件并不是一件小事。Kafka 上可能有延遲到達的數據，或者索引器將片段傳遞給歷史節點可能需要花些時間。為了解決這個問題，我們會在運行壓縮之前執行一些限制和檢查。

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

沒有這些措施，我們發現有時會丟失數據。在開始壓縮時仍有寫入的段將被新壓縮的段所覆蓋，這些段具有更高的版本，因此會優先。這可以有效地刪除包含在那些尚未完成傳遞的段中的數據。

4. 查詢

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

我們集群的大多數查詢都是由自定義的內部工具（如儀表板和預警系統）生成的。這些系統最初是為了與我們內部開發的開源時序數據庫 Atlas 一起工作而設計的。因此，這些工具使用 Atlas Stack 查詢語言。

為了加速查詢 Druid 的采用，并實現現有工具的重用，我們添加了一個翻譯層來接收 Atlas 查詢，將它們重寫為 Druid 查詢，發送查詢并將結果重新格式化為 Atlas 結果。這個抽象層允許現有的工具按原樣使用，用戶要訪問我們 Druid 數據存儲中的數據也不需要額外學習。

5. 調優

在調整集群節點的配置時，我們以較高的速度運行一系列可重復和可預測的查詢，從而獲得每個給定配置的響應時間和查詢吞吐量的基準。這些查詢在設計時隔離了集群的各個部分，以檢查查詢性能方面的改善或退化。

例如，我們對最近的數據進行有針對性的查詢，以便只對 Middle Manager 進行查詢。同樣，對于較長的時間段但較舊的數據，我們只查詢歷史節點來測試緩存配置。同樣，使用按高基數維分組的查詢檢查結果合并受到了什么影響。我們繼續調整和運行這些基準測試，直到我們對查詢性能滿意為止。

在這些測試中，我們發現調整緩沖區的大小、線程的數量、查詢隊列的長度和分配給查詢緩存的內存對查詢性能有實際的影響。然而，壓縮作業的引入對查詢性能有更重要的影響，它會將未充分匯總的段重新壓縮，實現完美匯總。

我們還發現，在歷史節點上啟用緩存非常有好處，而在代理節點上啟用緩存效果則不是很明顯。因此，我們不在代理上使用緩存。這可能是由我們的用例造成的，但是幾乎每一次查詢都會錯過代理上的緩存，這可能是因為查詢通常包含最新的數據，這些數據不在任何緩存中，因為一直有數據到達。

6. 小結

針對我們的用例和數據率，經過多次優化調整，Druid 已經被證明具備我們最初希望的能力。

我們已經能夠得到一個能力齊備、可用的系統，但仍然有更多的工作要做。隨著查詢數量和復雜性的增加，我們的攝取量和速率也在不斷增加。隨著更多的團隊認識到這些詳細數據的價值，我們經常需要添加更多的度量和維度，這加重了系統的負擔。我們必須繼續監控和調優，保證查詢性能受控。

目前，我們正在以每秒 200 萬次的速度處理事件，并查詢超過 1.5 萬億行的數據，以獲得關于用戶體驗服務的詳細信息。這一切幫助我們保持了高質量的 Netflix 體驗，同時能夠不斷地創新。