# 每天為 1000 億個事件賦予意義-Teads 的 Analytics（分析）管道 > 原文： [http://highscalability.com/blog/2018/4/9/give-meaning-to-100-billion-events-a-day-the-analytics-pipel.html](http://highscalability.com/blog/2018/4/9/give-meaning-to-100-billion-events-a-day-the-analytics-pipel.html) *這是 Teads.tv 的軟件工程師* *[Alban Perillat-Merceroz](https://www.linkedin.com/in/albanperillatmerceroz/?locale=en_US)* *的來賓帖子。*  在本文中，我們描述了如何將 Kafka，Dataflow 和 BigQuery 一起編排以攝取和轉換大量事件。 當添加規模和等待時間約束時，對它們進行協調和重新排序成為一個挑戰，這是我們如何解決的。 Teads for Publisher, one of the webapps powered by Analytics 在數字廣告中，日常運營會產生許多我們需要跟蹤的事件，以便透明地報告廣告系列的效果。 這些事件來自： * 用戶與與瀏覽器發送的廣告的互動。 這些事件稱為跟蹤事件，可以是標準事件（開始，完成，暫停，繼續等），也可以是自定義事件，這些事件來自使用 [Teads Studio](https://teads.tv/studio/) 構建的交互式廣告素材。 我們每天大約收到 100 億個跟蹤事件。 * 來自后端的事件大部分與廣告拍賣的詳細信息有關（實時出價過程）。 在采樣之前，我們每天會產生超過 600 億個此類事件，并且應當在 2018 年使這一數字增加一倍。 在文章中，我們將重點放在跟蹤事件上，因為它們是我們業務中最關鍵的路徑。 Simplified overview of our technical context with the two main event?sources 跟蹤事件由瀏覽器通過 HTTP 發送到專用組件，該組件除其他外將它們排入 [Kafka](https://kafka.apache.org/) 主題。 Analytics 是這些事件的使用者之一（更多內容請參見下文）。 我們有一個分析團隊，其任務是處理這些事件，其定義如下： > 我們攝取了越來越多的原木 > 我們將它們轉換為面向業務的數據， > 我們為每個觀眾提供高效且量身定制的服務。 為了完成此任務，我們構建并維護了一組處理工具和管道。 由于公司的有機增長和對新產品的要求，我們經常挑戰我們的體系結構。 ## 為什么我們搬到 BigQuery 早在 2016 年，我們的 Analytics（分析）堆棧基于 [lambda 架構](http://lambda-architecture.net/)（Storm，Spark，Cassandra），但我們遇到了幾個問題： * 數據規模使得不可能將所有數據都存儲在單個 Cassandra 表中，這阻止了有效的交叉查詢， * 這是一個復雜的基礎架構，具有用于批處理層和速度層的代碼重復，從而阻止了我們有效地發布新功能， * 最后，難以擴展且成本效益不高， 當時我們有幾種可能的選擇。 首先，我們可以構建一個增強的 lambda，但這只會推遲我們面臨的問題。 我們考慮了幾種有前途的替代方案，例如 [Druid](http://druid.io/druid.html) 和 [BigQuery](https://cloud.google.com/bigquery/) 。 我們最終因其強大的功能而選擇遷移到 BigQuery 。 使用 BigQuery，我們能夠： * 處理原始事件， * 使用 SQL 作為一種有效的數據處理語言， * 使用 BigQuery 作為處理引擎， * 使對日期的解釋性訪問更容易（與 Spark SQL 或 Hive 相比）， 多虧了 [固定費用計劃](https://cloud.google.com/bigquery/pricing#flat_rate_pricing) ，我們密集的使用（在查詢和存儲方面）具有成本效益。 但是，我們的技術背景對于 BigQuery 而言并不理想。 我們想使用它來存儲和轉換來自多個 Kafka 主題的所有事件。 我們無法將 Kafka 群集移出 AWS 或使用 [Pub / Sub](https://cloud.google.com/pubsub/docs/overview) （在 GCP 上與 Kafka 托管的等效版本），因為這些群集也由我們托管在 AWS 上的某些廣告投放組件使用。 結果，我們不得不應對運行多云基礎架構的挑戰。 今天， BigQuery 是我們的數據倉庫系統，在這里我們的跟蹤事件與其他數據源保持一致。 ## 攝取 在處理跟蹤事件時，您面臨的第一個問題是您必須無序處理，并且延遲未知。 事件實際發生的時間（事件時間）與系統觀察到事件的時間（處理時間）之間的差為毫秒至幾小時。 這些大的延遲并不是那么罕見，當用戶在兩次瀏覽會話之間失去連接或激活飛行模式時可能會發生。 Difference between event time and processing time 有關流數據處理挑戰的更多信息，建議您看一下 Google Cloud Next '17 演講? [在批處理和流處理之間來回移動](https://www.youtube.com/watch?v=PGTSZvBK8-Y) ?，來自 Tyler Akidau （Google 的數據處理技術負責人）和 Lo?cJaures（Teads 的聯合創始人兼 SVP Technology）。 本文受此演講啟發。 ## 流媒體的艱難現實 [數據流](https://cloud.google.com/dataflow/?hl=en)是一種托管流系統，旨在解決事件的混亂性質，以解決我們面臨的挑戰。 Dataflow 具有統一的流和批處理編程模型，流是旗艦功能。 我們按照 Dataflow 的承諾被出售，并坦率地嘗試了流模式。 不幸的是，在將其投入實際生產流量后，我們意外地感到驚訝： BigQuery 的流媒體插入成本。 我們是根據壓縮數據大小（即通過網絡的實際字節數）而不是 BigQuery 的原始數據格式大小來估算的。 幸運的是，現在已記錄在中，用于[每種數據類型](https://cloud.google.com/bigquery/pricing#data)。因此您可以進行數學計算。 那時，我們低估了這筆額外費用 100 倍，這幾乎使我們整個提取流程（Dataflow + BigQuery）的費用增加了一倍。 我們還面臨其他限制，例如 [100,000 個事件/秒速率限制](https://cloud.google.com/bigquery/quotas#streaminginserts)，這很危險地接近我們所做的事情。 好消息是，有一種方法可以完全避免流插入限制：批量加載到 BigQuery 中。 理想情況下，我們希望將數據流以流方式與 BigQuery 以批處理方式一起使用。 那時，Dataflow SDK 中沒有沒有 BigQuery 批處理接收器用于無限制的數據流。 然后，我們考慮開發自己的自定義接收器。 不幸的是，當時無法向無界數據流添加自定義接收器（請參閱 [Dataflow 計劃添加對在未來版本](https://cloud.google.com/dataflow/model/custom-io)中寫入無界數據的自定義接收器的支持-BeamBeam 現在有可能 是官方的 Dataflow SDK）。 我們別無選擇，只能將我們的數據流作業切換到批處理模式。 多虧了 Dataflow 的統一模型，僅需幾行代碼即可。 對我們來說幸運的是，我們能夠承受因切換到批處理模式而引入的額外數據處理延遲。 展望未來，我們當前的攝取架構基于 [Scio](https://github.com/spotify/scio) ，這是 Spotify 開源的用于數據流的 Scala API。 如前所述，Dataflow 本機支持 Pub / Sub，但 Kafka 集成還不成熟。 我們必須擴展 Scio 以實現偏移量檢查點持久性并實現有效的并行性。 ## 微量批次管道 我們得到的架構是一個由 30 分鐘的 Dataflow 批處理作業組成的鏈，這些作業按順序安排為讀取 Kafka 主題并使用加載作業寫入 BigQuery。 Phases of a Dataflow micro?batch 關鍵之一是找到理想批次持續時間。 我們發現在成本和讀取性能（因此延遲）之間具有最佳折衷的最佳選擇。 要調整的變量是 Kafka 讀取階段的持續時間。 為了獲得完整的批處理持續時間，您必須將寫入操作添加到 BigQuery 階段（不成比例，但與讀取持續時間緊密相關），并添加一個常數，即引導和關閉持續時間。 值得注意的是： * 讀取階段太短會降低讀取階段和非讀取階段之間的比率。 在理想的世界中，1：1 的比例意味著您必須能夠像寫作一樣快地閱讀。 在上面的示例中，我們有一個 20 分鐘的讀取階段，一個 30 分鐘的批處理（比率為 3：2）。 這意味著我們必須能夠比寫入速度快 1.5 倍。 較小的比率意味著需要更大的實例。 * 太長的讀取階段只會增加事件發生到 BigQuery 可用之間的延遲。 ## 性能調優 數據流作業是按順序啟動的，這是出于簡單原因和更容易進行故障管理。 這是我們愿意采取的延遲權衡。 如果作業失敗，我們只需返回到最后提交的 Kafka 偏移量即可。 我們必須修改 Kafka 群集的拓撲，并增加分區數量，以便能夠更快地對郵件進行堆棧。 根據您在數據流中進行的轉換，限制因素很可能是處理能力或網絡吞吐量。 為了實現高效的并行性，您應始終嘗試保持一定數量的 CPU 線程數作為分區數量的除數（推論：最好有很多[數量多的 Kafka 分區 復合數字](https://en.wikipedia.org/wiki/Highly_composite_number)）。 在極少的延遲情況下，我們可以使用更長的讀取序列來微調作業。 通過使用更大的批次，我們還能夠以延遲為代價來趕上延遲。 為了處理大多數情況，我們將 Dataflow 的大小設置為能夠讀取的速度比實際速度快 3 倍。 使用單個 [*n1-highcpu-16*](https://cloud.google.com/compute/docs/machine-types) 實例讀取 20 分鐘可以釋放 60 分鐘的郵件。 Ingestion latency (minutes) over?time 在我們的用例中，我們以鋸齒延遲結束，該鋸齒延遲在 3 分鐘（Write BQ 階段的最小持續時間）和 30 分鐘（作業的總持續時間）之間振蕩。 ## 轉型 原始數據不可避免地會龐大，我們有太多事件，無法按原樣查詢。 我們需要匯總這些原始數據以保持較低的*讀取時間*和緊湊的卷。 這是我們在 BigQuery 中的處理方式： Architecture overview spanning between AWS and?GCP 與傳統的 [ETL 流程](https://en.wikipedia.org/wiki/Extract,_transform,_load)之前的數據*轉換*之前，*加載*之前，我們選擇先將其（ELT）原始存儲 格式。 它有兩個主要優點： * 它使我們能夠訪問每個原始事件，以進行精細的分析和調試， * 通過讓 BigQuery 使用簡單但功能強大的 SQL 方言進行轉換，它簡化了整個鏈。 我們本來希望直接寫入每天分區的*原始事件*表。 我們之所以無法這樣做，是因為必須使用特定的目的地（表或分區）定義數據流批處理，并且其中可能包含用于不同分區的數據。 我們通過將每個批次加載到臨時表中然后開始對其進行轉換來解決此問題。 對于這些臨時批處理表中的每一個，我們運行一組轉換，具體化為輸出到其他表的 SQL 查詢。 這些轉換之一只是將所有數據追加到按天劃分的大型原始事件表中。 這些轉換的另一個是匯總：給定一組維度的數據匯總。 所有這些轉換都是冪等的，可以在發生錯誤或需要數據重新處理的情況下安全地重新運行 。 ## 匯總 直接查詢原始事件表對于調試和深度分析非常有用，但是查詢這種規模的表不可能獲得可接受的性能，更不用說這種操作的[成本](https://cloud.google.com/bigquery/pricing)了。 為了讓您有個想法，此表僅保留 4 個月，包含 1 萬億個事件，大小接近 250TB 。 Example of a rollup transformation 在上面的示例中，我們匯總了 3 個維度的事件計數：*小時*，*廣告 ID* 和*網站 ID* 。 事件也被透視并轉換為列。 該示例顯示尺寸縮小了 2.5 倍，而實際情況更接近 70 倍。 在 BigQuery 的大規模并行上下文中，查詢運行時的影響不大，改進程度取決于所使用的廣告位數量。 匯總還使我們可以將數據劃分為小塊：在給定的小時（事件時間的小時，而不是處理時間）中，事件被分組到小表中。 因此，如果您需要查詢給定時間的數據，則將查詢一個表（< 10M 行，< 10GB）。 匯總是一種通用匯總，我們可以在給定大量維度的情況下更高效地查詢所有事件。 在其他一些用例中，我們需要專用的數據視圖。 他們每個人都可以實施一組特定的轉換，以最終得到一個專門的優化表。 ## 托管服務的限制 BigQuery 功能強大，但有其局限性： * BigQuery 不允許查詢具有[不同架構](https://cloud.google.com/bigquery/docs/querying-wildcard-tables#schema_used_for_query_evaluation)的多個表（即使查詢未使用不同的字段）。 當需要添加字段時，我們有一個腳本可以批量更新數百個表。 * BigQuery [不支持列刪除](https://stackoverflow.com/a/45822880)。 沒什么大不了的，但無助于償還技術債務。 * 查詢多個小時：BigQuery [支持表名](https://cloud.google.com/bigquery/docs/querying-wildcard-tables#best_practices)中的通配符，但是性能太差了，我們必須生成查詢以 UNION ALL 顯式查詢每個表。 * 我們始終需要將這些事件與托管在其他數據庫上的數據結合起來（例如，將事件與廣告活動的更多信息結合在一起），但是 BigQuery 不支持（目前）。 我們目前必須定期將整個表復制到 BigQuery，以便能夠在單個查詢中聯接數據。 ## 云間數據傳輸的樂趣 借助 AWS 中 Teads 的廣告投放基礎設施以及與許多其他組件共享的 Kafka 集群，我們別無選擇，只能在 AWS 和 GCP 云之間移動大量數據，這并不容易，而且肯定不會 賤。 我們將 Dataflow 實例（因此是主要 GCP 入口點）放置在離我們 AWS 基礎設施盡可能近的位置，幸運的是，AWS 和 GCP 之間的現有鏈接足夠好，因此我們可以簡單地使用托管 VPN。 盡管我們在運行這些 VPN 時遇到了一些不穩定因素，但是我們還是設法使用一個簡單的腳本將其關閉然后重新打開，以進行分類。 我們從來沒有遇到足夠大的問題來證明專用鏈接的成本合理。 再一次，[成本是您必須密切注意的](https://medium.com/teads-engineering/real-life-aws-cost-optimization-strategy-at-teads-135268b0860f)成本，并且就出口而言，很難在看到賬單之前進行評估。 仔細選擇壓縮數據的方式是降低這些成本的最佳方法之一。 ## 只有一半 Teads’ Analytics big?picture 在 BigQuery 中僅包含所有這些事件還不夠。 為了為業務帶來價值，必須使用不同的規則和指標對數據進行合并。 此外，BigQuery 并非針對實時用例量身定制。 由于并發限制和 3 到 5 秒鐘的不可壓縮查詢延遲（這是其設計的可接受和固有的），因此必須將 BigQuery 與其他工具（服務于儀表板，Web UI 等）組合使用。 此任務由我們的 Analytics 服務執行，這是一個 Scala 組件，可利用 BigQuery 生成按需報告（電子表格）和量身定制的數據集市（每天或每小時更新）。 需要此特定服務來處理業務邏輯。 否則很難維護為 SQL 并使用管道轉換生成數據集市。 我們選擇了 AWS [Redshift](https://aws.amazon.com/redshift/) 來存儲和服務我們的數據集市。 盡管服務面向用戶的應用似乎不是一個顯而易見的選擇，但是 Redshift 對我們有用，因為我們的并發用戶數量有限。 同樣，使用鍵/值存儲將需要更多的開發工作。 通過保留中間關系數據庫，可以簡化數據集市的使用。 在上有很多話要說，我們如何以規模構建，維護和查詢這些數據集市，但這將是另一篇文章的主題。 * * * 如果您喜歡，大規模事件處理和與 Analytics（分析）相關的挑戰，請大聲疾呼。 我們正在積極[尋找](https://teads.tv/teads-jobs/)，以供工程師構建未來的架構。