摘要：本文由 Netflix 進階軟體工程師徐振中分享，内容包含有趣的案例、分布式系統基礎方面的各種挑戰以及解決方案，此外還讨論了其在開發運維過程中的收獲，對開放式自助式實時資料平台的一些新願景，以及對 Realtime ETL 基礎平台的一些新思考。文章内容主要分為以下三部分：
- 産品背景
- 産品功能
- 挑戰&解決方案
Netflix 緻力于會員的喜悅。我們不懈地專注于提高産品體驗和高品質内容。近年來，我們一直在技術驅動的 Studio 和内容制作方面進行大量投資。在這個過程中，我們發現在實時資料平台的領域裡中出現了許多獨特并有意思的挑戰。例如，在微服務架構中，領域對象分布在不同的 App 及其有狀态存儲中，這使得低延遲高一緻性的實時報告和實體搜尋發現特别具有挑戰性。
Netflix 的長久願景是把歡樂和微笑帶給整個世界，通過在全球各地拍攝一些高品質、多元化的内容産品放在平台上，分享給平台超過一個億級别的使用者。更遠大的目标為了給使用者帶來愉悅的體驗，Netflix 的努力方向分為兩個：
- 一方面是通過資料整合知識來回報并用于提高使用者的産品體驗中去；
- 另一方面通過建立一個技術驅動的 Studio 去幫助産出内容品質更高的産品。
而作為一個資料平台團隊，需要關注的是怎麼幫助公司中不同的開發人員、資料分析人員等實作其在公司中的價值，最終為解決上述兩方面問題做出自己的貢獻。
簡單地介紹一下 Netflix 資料平台團隊及相應的産品，Keystone。它的主要功能是幫助公司在所有的微服務中埋點、建立 Agent、釋出事件、收集事件資訊，然後放到不同的資料倉庫中進行存儲，比如 Hive 或 ElasticSearch，最後幫助使用者在資料實時存儲的情況下實作計算和分析。
- 從使用者的角度來講，Keystone 是一個完整的自容（Self-contained）的平台,支援多使用者，使用者可以通過所提供的 UI 很友善地聲明并建立自己想要的 pipeline。
- 從平台角度來說，Keystone 提供底層所有分布式系統中實作比較困難的解決方案，如容器編排（Container Orchestration）、工作流管理（Workflow Management）等等，這些對于使用者是不可見的。
- 從産品的角度來說，主要有兩個功能，一個是幫助使用者将資料從邊緣裝置移到數倉，另一個是幫助使用者實時計算的功能。
- 從數字的角度來說，Keystone 在 Netflix 的使用是非常有必要的，隻要跟資料打交道的開發者，一定會用到，是以 Keystone 在整個公司中有幾千個使用者，并有一百個 Kafka 的叢集支援每天 10PB 數量級左右的資料。
Keystone 的整個架構分為兩層，底層是 Kafka 和 Flink 作為底層引擎，底層對所有分布式系統中比較困難的技術方案進行抽象，對使用者不可見，在上層建構整個應用；服務層會提供抽象的服務，UI 對于使用者來講比較簡單，不需要關心底層實作。
下面介紹一下 Keystone 産品在過去四五年的發展曆程。最初的動機是收集所有裝置的資料并将其存儲到資料倉庫中，當時使用的是 Kafka 技術，因為資料移動比較好解決，本質上來講僅是一個多并發的問題。
在此之後，使用者給出了新的需求，即在資料移動的過程中對資料進行一些簡單的處理操作，比如篩選（Filter），還有一個很通用的功能 —— projection，為此 Keystone 推出了針對該需求推出了相應的功能特性。
經過一段時間後，使用者表示想做更加複雜的 ETL，比如 Streaming Join 等，是以産品決定将底層的 API 提供給使用者，并将底層的關于所有分布式系統的解決方案抽象化，讓其更好地關注上層的内容。
## 産品功能
産品功能介紹将圍繞 Netflix 中的兩個“超級英雄” Elliot 和 Charlie 來展開。Elliot 是來自資料科學工程組織的一個資料科學家，他的需求是在非常大的資料中尋找響應的 pattern，以幫助提高使用者體驗；Charlie 是一個來自 Studio 組織的應用開發者，其目标是通過開發一系列的應用來幫助周邊的其他開發者産出更高品質的産品。
這兩個人的工作對于産品來講都非常重要，Elliot 的資料分析結果可以幫助給出更好的推薦和個性化定制,最終提高使用者體驗；而 Charlie 的工作可以幫助周邊的開發者提高效率。
Recommendation & Personalization
Elliot 作為一個資料科學家，需要的是一個簡單易用的實時 ETL 操作平台，他不希望寫非常複雜的編碼，同時需要保證整個 pipeline 的低延時。他所從事的工作和相關需求主要有以下幾個：
- 推薦和個性化定制。該工作中可以根據個人特點的不同将同樣的視訊通過不同的形式推送給相應的使用者，視訊可以分為多個 row，每一個 row 可以是不同的分類，根據個人的喜好可以對不同的 row 進行更改。此外，每一個視訊的題目都會有一個 artwork，不同國家、不同地域的不同使用者對 artwork 的喜好也可能不同，也會通過算法進行計算并定制适合使用者的 artwork。
- A/B Testing。Netflix 提供給非會員使用者 28 天免費的視訊觀看機會，同時也相信給使用者看到了适合自己的視訊，使用者更有可能會購買 Netflix 的服務，而在進行A/B Testing 的時候，就需要 28 天才能做完。對于 Elliot 來講，進行 A/B Testing 的時候可能會犯錯誤，他所關心的是怎麼樣才能在不用等到 28 天結束的時候就可以提前發現問題。
當在裝置上觀看 Netflix 的時候，會以請求的形式和網關進行互動，然後網關會将這些請求分發給後端的微服務，比如說使用者在裝置上點選播放、暫停、快進、快退等操作，這些會有不同的微服務進行處理，是以需要将相應的資料收集起來，進一步處理。
對于 Keystone 平台團隊來講，需要收集不同的微服務中産生的資料并進行存儲。Elliot 需要将不同的資料整合起來，以解決他關注的問題。
至于為什麼要使用流處理，主要有四方面的考量，即實時報告、實時告警、機器學習模型的快速訓練以及資源效率。相比于前兩點，機器學習模型的快速訓練以及資源效率對 Elliot 的工作更加重要。尤其需要強調的是資源效率，針對前面的 28 天的 A/B Testing，目前的做法是每天将資料與前 27 天做 Batch Processing，這個過程中涉及了很多重複處理，使用流處理可以很好地幫助提高整體的效率。
Keystone 會提供指令行的工具給使用者，使用者隻需要在指令行中輸入相應的指令來進行操作，工具最開始會詢問使用者一些簡單的問題，如需要使用什麼 repository 等，使用者給出相應的回答後，會最終産生一個模闆，使用者便可以開始使用工具進行開發工作；産品還提供一系列簡單的 SDK，目前支援的是Hive、Iceberg、Kafka 和 ElasticSearch 等。
需要強調的是 Iceberg，它是在 Netflix 主導的一個 Table Format，未來計劃取代 Hive。其提供了很多特色功能來幫助使用者做優化；Keystone 向使用者提供了簡單的 API，可以幫助其直接生成 Source 和 Sink。
Elliot 在完成一系列的工作之後，可以選擇将自己的代碼送出到 repository 中，背景會自動啟動一個 CI/CD pipeline，将所有的源代碼和制品等包裝在 Docker 鏡像中，保證所有的版本一緻性。Elliot 在 UI 處隻需要選擇想要部署哪一個版本，然後點選部署按鈕可以将 jar 部署到生産環境中。
産品會在背景幫助其解決底層分布式系統比較困難的問題，比如怎麼做容器編排等，目前是基于資源的編排，未來計劃向 K8S 方向發展。部署 Job（作業）包的過程中會部署一個 JobManager 的叢集和一個 TaskManager 的叢集，是以每一個 Job 對于使用者來說是完全獨立的。
産品提供預設的配置選項，同時也支援使用者在平台 UI 上修改并覆寫配置資訊，直接選擇部署即可生效，而不需重寫代碼。Elliot 之前有一個需求是在 Stream Processing 的過程中，比如從不同的 Topic 中去讀取資料，出現問題的情況下可能需要在 Kafka 中操作，也可能需要在資料倉庫中操作，面對該問題，其需求是在不改動代碼的情況下切換不同的 Source，而目前平台提供的UI很友善地完成該需求。此外平台還可以幫助使用者在部署的時候選擇需要多少資源來運作作業。
很多使用者從 Batch Processing 轉到 Stream Processing 的過程中，已經有了很多需要的制品，比如 Schema 等，是以平台還幫助其簡單地實作這些制品的內建。
平台擁有很多需要在其之上寫 ETL 工程的使用者，當使用者越來越多的時候，平台的可伸縮性顯得尤為重要。為此，平台采用了一系列的 pattern 來解決該問題。具體來講，主要有三個 pattern 正在使用，即 Extractor Pattern、Join Pattern 和 Enrichment Pattern。
Content Production
先簡要介紹一下什麼是 Content Production。包括預測在視訊制作方面的花費、制定 program、達成 deal、制作視訊、視訊後期處理、釋出視訊以及金融報告。
Charlie 所在的是 Studio 部門主要負責開發一系列的應用來幫助支援 Content Production。每一個應用都是基于微服務架構來開發部署的，每一個微服務應用會有自己的職責。舉個最簡單的例子，會有專門管理電影标題的微服務應用，也會有專門管理 deals 和 contracts 的微服務應用等等。
面對如此多的微服務應用，Charlie 面臨的挑戰問題是當其在進行實時搜尋的過程中，比如搜尋某一個電影的演員，需要将資料從不同的地方 join 起來；另外資料每天都在增加，保證明時更新的資料的一緻性比較困難，這本質上是分布式微服務系統的特點導緻，不同的微服務選擇使用的資料庫可能不同，這給資料一緻性的保證又增加了一定的複雜度。針對該問題，常用的解決方案有以下三個：
- Dual writes: 當開發者知道資料需要放到主要的資料庫中的時候，同時也要放到另一個資料庫中，可以很簡單地選擇分兩次寫入到資料庫中，但是這種操作是不容錯的，一旦發生錯誤，很有可能會導緻資料的不一緻；
- Change Data Table: 需要資料庫支援事務的概念，不管對資料庫做什麼操作，相應的變更會加到事務變更的 statement 中并存入單獨的表中，之後可以查詢該 change 表并擷取相應的變更情況并同步到其他資料表；
- Distributed Transaction：指的是分布式事務，在多資料環境中實作起來比較複雜。
Charlie 的一個需求是将所有的電影從 Movie Datastore 複制到一個以 Elasticsearch 來支援的 movie search index 中，主要通過一個 Polling System 來做資料拉取和複制，資料一緻性的保證采用的是上述的 Change data table 的方法。
該方案的弊端是隻支援定期資料拉取，另外 Polling System 和資料源直接緊密結合，一旦 Movie Search Datastore 的 Schema 改變，Polling System 就需要修改。為此，該架構在後來做了一次改進，引入了事件驅動的機制，讀取資料庫中所有實作的事務，通過 stream processing 的方式傳遞到下一個 job 進行處理。為了普适化該解決方案，在 source 端實作了不同資料庫的 CDC（Change Data Capture）支援，包括 MySQL、PostgreSQL 和 Cassandra 等在 Netflix 中比較常用的資料庫，通過 Keystone 的 pipeline 進行處理。
挑戰及解決方案
下面分享一下上述方案存在的挑戰和相應的解決方案：
- Ordering Semantics
在變更資料事件中，必須要保證 Event ordering，比如一個事件包含 create、update 和 delete 是三個操作，需要傳回給消費者側一個嚴格遵守該順序的操作事件。一個解決方案是通過 Kafka 來控制；另一個解決方案是在分布式系統中保證捕獲的事件與實際從資料庫中讀取資料的順序是一緻的，該方案中當所有的變更事件捕獲出來後，會存在重複和亂序的情況，會通過 Flink 進行去重和重新排序。
- Processing Contracts
在寫 stream processing 的時候，很多情況下不知道 Schema 的具體資訊，是以需要在消息上定義一個契約 contract，包括 Wire Format 以及在不同的層級上定義與Schema 相關的資訊，如基礎設施（Infrastructure）、平台（Platform）等。Processor Contract 的目的是幫助使用者将不同的 processor metadata 組合起來，盡量減少其寫重複代碼的可能。
舉一個具體的案例，比如 Charlie 希望有新的 deal 的時候被及時通知，平台通過将相關的不同元件組合起來，DB Connector、Filter 等，通過使用者定義契約的方式幫助其實作一個開放的可組合的流資料平台。
以往所看到的 ETL 工程大多數适用于資料工程師或資料科學家。但從經驗上來講，ETL 的整個過程，即 Extract、Transform 和 Load，其實是有被更廣泛應用的可能。最早的 Keystone 簡單易用，但靈活性不高，之後的發展過程中雖然提高了靈活性，但複雜性也相應地增大了。是以未來團隊計劃在目前的基礎上進一步優化，推出開放的、合作的、可組合的、可配置的 ETL 工程平台，幫助使用者在極短的時間解決問題。
作者簡介：
徐振中，Netflix 軟體工程師，在 Netflix 從事高度可擴充和彈性的流媒體資料平台的基礎設施工作，熱衷于研究分享與實時資料系統、分布式系統基本原理相關的任何有趣的事情！