# TripleLift 如何建立 Adtech 數據管道每天處理數十億個事件 > 原文： [http://highscalability.com/blog/2020/6/15/how-triplelift-built-an-adtech-data-pipeline-processing-bill.html](http://highscalability.com/blog/2020/6/15/how-triplelift-built-an-adtech-data-pipeline-processing-bill.html)  *這是 [TripleLift](https://triplelift.com/) 的數據工程師 [Eunice Do](https://www.linkedin.com/in/eunicedo/) 的來賓帖子，該公司領導下一代程序化廣告。* ## 您的系統名稱是什么，我們在哪里可以找到更多信息？ 該系統是 TripleLift 上的數據管道。 TripleLift 是一家廣告技術公司，與該行業的大多數公司一樣，我們每天都要處理大量數據。 在過去的三年中，TripleLift 數據管道的規模從每天處理數百萬個事件擴展到處理數十億個事件。 可以將這種處理總結為以經濟高效的方式將報告數據連續聚合和傳遞給用戶。 在本文中，我們將主要關注這一數十億事件管道的當前狀態。 要跟蹤到目前狀態的 5 年歷程，請查看[，這是我們的工程副總裁關于 TripleLift 管道](https://www.datacouncil.ai/talks/the-highs-and-lows-of-building-an-adtech-data-pipeline)的演變的演講。 ## 您為什么決定構建此系統？ 我們需要一個可以實現以下目的的系統： * 隨著數據量的快速增長而擴展 * 以經濟高效的方式匯總日志級別的數據 * 擁有清晰，可管理的工作依賴鏈 * 自動化冪等作業運行和重試 * 在預期的 SLA 中將數據交付給 BI 和報告工具 * 處理那些報表工具上的查詢負載 ## 您的系統有多大？ 嘗試感受一下系統的工作量。 大概系統統計-2020 年 4 月左右 * 最大的事件日志每天接收 300 億個事件 * 5 個日常聚合作業，最大輸出為 7.5 GB * 每小時 25 個聚合作業，最大輸出 2.5 GB * 15 小時的工作將匯總數據導入 BI 工具 * 基數最高的歸一化聚合具有 75 個維度和 55 個指標 ## 您的輸入/輸出帶寬使用量是多少？ 數據管道匯總了我們 Kafka 集群收集的事件數據，其 I / O 約為 2.5GB / hr。 ## 您的成長速度有多快？ 在過去的幾年中，我們經歷了偶然的，快速的同比增長。 從 2016 年到 2017 年，我們管道中處理的數據量增長了 4.75 倍。 然后是 2017 年至 2018 年的 2.5 倍。最后是 2018 年至 2019 年的 3.75 倍。這就是 **3 年后的近 50 倍**！ ## 您的系統架構如何？ 在較高級別，我們的數據管道運行批處理，其流程包括： 1. 原始事件收集和持久性 2. 通過多個聚合級別對數據進行非規范化和規范化 3. 將聚合數據持久性或攝取到各種數據存儲中 4. 在用戶界面和報告工具中展示提取的數據以進行查詢 我們從數據收集過程開始，在該過程中，將原始事件數據發布到我們的 50 多個 **Kafka** 主題中。 這些事件由 **Secor** （由 Pinterest 創建的開源使用者）使用，并以實木復合地板格式寫到 **AWS S3** 。 我們使用 **Apache Airflow** 來促進步驟 2 和 3 中必需的調度和依賴關系管理。 Airflow 通過向 Databricks API 的作業提交請求啟動聚合任務。 聚合使用 **Databricks** 群集上的 **Apache Spark** 運行。 首先，通過加入大量原始事件日志將數據歸一化為寬表，以完整描繪廣告位的拍賣前，拍賣中和拍賣后的情況。 非規范化日志將保留到 S3。 在非標準化任務進入成功狀態之后，Airflow 調度程序將啟動其下游標準化任務。 這些聚合中的每一個都會將非規范化數據匯總到更窄范圍的維度和指標集中，以適合特定報表的業務環境。 這些最終聚合也將保留到 S3。 成功完成每個最終聚合任務后，Airflow 調度程序將啟動其各種下游持久性或攝取任務。 其中一項任務是將匯總數據復制到 **Snowflake** 中，該數據分析平臺用作我們的商業智能工具的后端。 另一個任務將數據提取到 **Imply Druid** 中，這是一種托管的云解決方案，由時間優化的列式數據存儲組成，支持對大型數據集的即席分析查詢。 最后，第 4 步是我們的商業智能和數據工程團隊之間的共同努力。 可以查詢聚合數據的主要地方是我們的內部報告 API， **Looker** （由 Snowflake 支持）和 **Imply Pivot** （拖放分析 UI 捆綁在 Imply 中 德魯伊解決方案）。 ## 您學到了什么？ 數據決策往往會產生深遠的影響。 例如： * 一旦定義了現有字段，就很難改變其衍生方式。 這是由于維護該字段的歷史連續性的共同需要。 * 如果多個聚合級別中的任何一個級別的錯誤都持續運行了一段時間，那么用不正確的數據回填該時間段可能會既耗時又昂貴。 * 如果提供了不正確或不完整的數據以供查詢，則不會告訴您該數據在何處結束。 ## 您希望自己做些什么？ 長期以來，我們對于可訪問的數據范圍或可訪問的數據范圍尚無明確的方法。 * 不存在保留策略，并且無限期地存儲了許多數據。 * 允許用戶查詢無限期存儲的數據，這降低了其他所有人的查詢性能。 * 所有報告工具均支持所有類型的查詢。 換句話說，我們無法正確識別內部報告與外部報告以及報告與分析用例。 由于我們缺乏使數據可訪問性的紀律性，最終形成了數據。 例如，我們沒有熱對冷分層，高對低粒度報告數據或合理的數據保留策略。 此后，我們對方法進行了更多思考，并采取了一些措施，例如實施 AWS S3 生命周期規則，為每個可查詢數據源定義保留策略，以及指定報告工具以處理快速的調查性查詢或較長日期范圍內的大型報告，但 都。 ## 您如何考慮將來更改架構？ 我們計劃通過使用 **Kafka Streams** 構建實時流式應用程序來補充批處理管道。 這是從涉及 **Spark 結構化流**， **Kafka 流**和 **KSQL** 的一些概念證明中選擇的。 ## 您如何繪制網絡和服務器統計數據及趨勢圖？ 我們使用 **Prometheus** 存儲應用程序指標。 然后，我們在 **Grafana 儀表板**中匯總指標，并在這些儀表板上設置警報。 ## 您的團隊中有多少人？ 團隊中共有 **4 位數據工程師**，而我們約占整個工程組織的十分之一。 我們經常與之合作的團隊是基礎架構，解決方案和數據科學。 例如，我們最近通過 Airflow 加入了數據科學團隊，并且他們的模型運行現在是自動化的。 ## 您使用哪種語言來開發系統？ **用于氣流的 Python** ，用于聚合的 Spark Scala 和用于某些報告工具的 Java。