# 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。