用近乎实时的分析来衡量Uber货运公司的指标

IT大咖说

发布于 2022-09-28 15:40:46

5300

发布于 2022-09-28 15:40:46

◆ 简介

虽然大多数人都熟悉Uber，但并非所有人都熟悉优步货运, 自2016年以来一直致力于提供一个平台，将托运人与承运人无缝连接。我们正在简化卡车运输公司的生活，为承运人提供一个平台，使其能够浏览所有可用的货运机会，并通过点击一个按钮进行预订，同时使履行过程更加可扩展和高效。

为托运人提供可靠的服务是优步货运获得他们信任的关键。由于承运人的表现可能会大大影响货运公司服务的可靠性，我们需要对承运人透明，让他们知道我们对他们负责的程度，让他们清楚地了解他们的表现，如果需要，他们可以在哪些方面改进。

为了实现这一目标，优步货运公司开发了承运人记分卡，以显示承运人的几个指标，包括对应用程序的参与度、准时取货/交货、跟踪自动化和延迟取消。通过在承运人应用程序上近乎实时地显示这些信息，我们能够实时向承运人提供反馈，使我们在行业内的大多数竞争对手中脱颖而出。

◆ 如何做到

后台要求

在规模上快速访问新鲜数据是建立可扩展的运营商记分卡后端的关键要求。

数据的新鲜度--一旦负载完成或被退回（被承运人取消），性能分数就会以最少的延迟更新。
延迟 - 运营商能够在页面加载延迟较低的应用程序中查看其性能得分。
可靠性 - 数据可以被可靠地处理和提供，在系统故障或代码重新部署的情况下，服务可以被优雅地恢复。
准确性 - 绩效指标的计算必须准确，以避免过多的入境纠纷。

考虑的潜在解决方案

◆ 使用实时聚合 MySQL

优点能够处理高读/写量的可靠数据库能够支持近乎实时的数据，因为MySQL可以从流式数据源进行更新
弊端大数据集的复杂聚合查询并不像我们所希望的那样具有性能，而且随着我们的持续增长和数据规模的扩大，可能会使MySQL实例陷入瓶颈。我们将需要批量上载记录，以确保历史数据是最新的。频繁的大批量插入会降低飞行中SELECT查询的性能

◆ 预先聚合数据 MySQL

优点由于数据在加载到MySQL之前已经被预先汇总，所以查询非常高效，因为他们可以完全利用MySQL索引来拉动单一的数据行
弊端无法支持近乎实时的指标，因此需要按照既定的节奏分批进行预汇总。与上述解决方案类似，我们需要批量上载记录，以确保历史数据是最新的，这可能导致性能问题。我们需要预先聚合每一个必要的用例，当新的用例与聚合参数出现后，可能会增加维护服务的工作量。

◆ 两个OLTP数据库表

一个表存储原始事件，事件更新触发一个异步函数来更新另一个汇总表中的所有相关指标。

优点可用性高，查询延迟低，最终达到一致
弊端很难迁移事件模式和重新定义指标的逻辑，因为新的指标定义需要重新计算所有预先汇总的指标。可扩展性不强，尤其是在写作流量大的时候

◆ 用OLAP Apache Pinotᵀᴹ

Apache Pinot是一个实时、分布式和可扩展的数据存储，旨在以面向用户的分析所需的超低延迟执行分析查询。一个单一的逻辑表（又称混合表）可以被设置为实时和离线摄取，基于 lambda架构.

优点运营商的性能数据可以通过利用Apache Flink®和Kafka®进行实时测量。Apache Flink和Kafka®，然后这些性能指标事件可以被摄取到混合Pinot表的实时部分。来自HDFS的离线、清理过的数据源可用于填充混合Pinot表的离线部分，确保完成后对货物的任何修改都被考虑在内。Pinot提供了许多索引选项，以实现低延迟的聚合和数据选择查询
弊端业务逻辑需要在两个不同的地方维护。虽然我们已经在离线HDFS中维护了定义这些性能指标的逻辑，但我们也需要保持实时Flink逻辑的更新，以确保它们输出一致的数据。

◆ 最终系统设计

为了实现对实时数据的精确分析，我们决定使用Lambda架构，利用Kafka、Flink和Pinot。然后，一旦数据生成，我们就用Redis进行缓存，并通过Golang GRPC端点向我们的下游客户（运营商应用程序和网络平台）提供汇总的指标。

◆ Kafka

Apache Kafka是Uber技术栈的一个基石。我们拥有世界上最大的Kafka部署之一，并且做了大量有趣的工作来确保它的性能和可靠性。随着货运业务的增长，Kafka可以轻松地扩展。

◆ Flink

Apache Flink是一个开源框架，用于对数据流进行有状态的计算。在我们的用例中，这对于从其他后端服务消耗原始事件、过滤不相关的事件、将它们映射到持久化的状态、确定性能质量，以及输出到具有共同事件模式的Kafka主题是必要的。Flink可以处理非常大的流量，也有很好的容错能力。

◆ Pinot

Apache Pinot是一个开源的、分布式的、高度可扩展的OLAP数据存储，它为每秒有数千次并发查询的网络规模的应用提供低查询延迟（即P99延迟在几秒钟之内）。它支持对单个表进行SQL分析。它使用现在流行的Lambda数据架构，从实时流和批处理数据源摄取数据，用于历史数据。

在货运公司的用例中，Pinot使用来自Kafka的实时数据摄取来覆盖过去3天内创建的数据。对于历史数据，Pinot从HDFS摄取，以覆盖从3天前到时间开始的数据。离线摄取管道有内置的回填能力，可以在需要时对以前的数据进行修正。

Apache Pinot提供了丰富的索引优化技术，如倒置、星形树、JSON、排序列等。索引来加速查询性能。例如，在星型树预聚合索引可以加快查询速度，总结出设施的平均等待时间。快速的查询使承运人在预订货物之前，在承运人的应用程序上查看等待时间，这是一种互动体验。

◆ 数据模式

输出主题提供了一个一般模式，每个事件有一行。这使我们能够在未来添加额外的事件名称选项，以支持未来的需求。它还提供了基础层面的数据点，可用于汇总我们的记分卡所需的所有指标。

◆ 查询实例

下面是一个常用查询的例子，用于提取某一时间窗口内某一承运人完成的工作总数和行驶的里程。

过滤器条款中使用的值是根据客户提供的API请求输入而变化的。

◆ Flink有状态流处理器

◆ 数据来源

货运后端服务通过一个内部的事件聚合服务将事件数据输出到Kafka。从这个统一的事件流主题，我们可以将这些Kafka事件消费到我们的Flink流处理引擎中。来自这个主题的事件包括诸如预约时间变化、实际到达预约地点的时间、货物状态变化等等。

◆ 关键状态

一旦货件被承运人预订，就会为给定的货件UUID创建一个状态对象。这个货件UUID可以在将来用来检索当前的状态，并参考有关该货件的共同细节。

创建。在首次创建状态时，我们会调用其他后台服务来填充初始细节，如停靠地点、承运人和司机标识等，并将其填充到状态对象中。
更新。随着重要更新事件的处理，状态会被更新，以反映被改变的新货物细节。
删除。一旦运输完成，事件就会被最终确定，状态对象就会被删除。

◆ 阶段性成果

每当一个里程碑被击中，Kafka消息就会被输出到我们之前讨论的数据模式中的sink主题。里程碑的一个例子是我们的自动跟踪得分。如果一个站点被司机标记为 "到达"，而不是被我们的运营团队标记为 "到达"，那么auto_arrived_at_stop的值会被输出，布尔值为True。

◆ 挑战

◆ 模式的演变

为了能够重新启动作业，从上次离开的地方继续前进，Flink将创建检查点并将其存储在HDFS中。为了对键入的状态进行处理，状态对象被序列化，然后保存到检查点文件中。当工作重新启动时，状态会从最近的检查点加载，并且对象会被反序列化为Java实例。当我们试图在状态对象中添加一个新的字段时，问题就出现了。工作未能从检查点加载，因为序列化的对象无法被反序列化为新的对象实例。为了解决这个问题，我们利用了Apache AVROᵀᴹ来为状态对象定义一个模式。从这个模式中，AVRO生成的对象可以被安全地序列化和反序列化，即使字段被改变，只要这些改变遵守了模式演化规则.

◆ 内存分配的优化

当我们刚开始在staging中运行我们的Flink作业时，我们一直遇到内存问题，作业会崩溃。我们试图修改这些数值，但要确保我们的工作顺利运行，获得正确的配置并不是一个简单的过程。幸运的是，Uber的一位同事分享了一个关于如何正确配置这些内存设置的非常有用的演示。如果你有兴趣了解更多信息，可以找到该演示文稿这里如果你有兴趣了解更多。

◆ 混合Pinot表

对于每个混合Pinot表，在引擎盖下有两个物理表：一个用于实时数据，另一个用于离线历史数据。Pino头Broker通过执行离线和实时联合，确保实时表和离线表之间的重叠部分正好被查询到一次。

让我们来看看这个例子，我们有5天的实时数据--3月23日至3月27日，而离线数据已经推送到3月25日，这比实时数据晚了2天。经纪人维持这个时间界限。

假设，我们得到一个对这个表的查询：select sum(metric) from table。代理商将根据这个时间界限把这个查询分成两个查询--一个是离线查询，一个是实时查询。这个查询变成：select sum(metric) from table_REALTIME where date >= Mar 25 and select sum(metric) from table_OFFLINE where date < Mar 25。代理商在将结果返回给客户之前，会合并这两个查询的结果。

查询性能

下面的查询例子是针对Pinot表最常用的一个查询。该查询是一个分析性查询，有聚合、分组和过滤条款。目前的查询量约为每秒40次。在拥有10G数据的表中，Pinot能够提供~250ms的P99查询延迟。这种水平的查询性能为我们的货运应用用户提供了良好的互动体验。

◆ 表的优化

为了实现表的250ms查询延迟，我们在Pinot表上使用两种类型的索引。

颠倒的索引对于事件名称、承运人uuid、司机uuid、工作uuid、装载uuid、预订uuid、停止uuid、市场类型和预订渠道

倒置索引可以将WHERE子句中相应过滤条件的查询速度提高10倍。

排序的索引在离线摄取管道中按carrier_uuid排序，这使我们的表的大小减少了一半，从而降低了查询延迟。Pinot是一个列式存储，像load_uuid、job_uuid、carrier_uuid和driver_uuid这样的列可以在多条记录中具有相同的值，而数据是物理排序的，因此可以显著提高压缩率。carrier_uuid和event_name总是被用作查询中的过滤器，对它们进行物理排序可以减少查询时加载的片段数量。

◆ Golang GRPC服务

Neutrino是一个主要的查询网关，用于访问Uber的Pinot数据集。它是一个稍微不同的部署胜利者在这里，每个主机上都运行着一个协调器和一个工作者，并且能够独立运行每个查询。Neutrino是一个托管在Mesos容器上的无状态和可扩展的常规Java微服务。它接受PrestoSQL查询，将其翻译成Pinot查询语言，并将其路由到适当的Pinot集群。本机和Neutrino Presto的主要区别是在于，Neutrino做了积极的查询推送，以最大化底层存储引擎的利用率。

◆ 缓存

当用户在移动应用中打开或刷新运营商记分卡时，将同时获取5个指标，这相当于9个Neutrino查询，因为有些指标需要超过一个Neutrino查询。我们的Neutrino查询的P99延迟约为60ms，为了减少Neutrino的流量并改善外部延迟，我们在Neutrino前面添加了一个Redis缓存，用来存储聚合的指标。设置了12小时的TTL，随着新事件的不断涌入，我们使用以下策略来确保缓存的一致性。平均而言，我们能够实现>90%的缓存命中率。

缓存之外。当请求键在读取过程中尚未被缓存（缓存缺失）时，我们会查询Neutrino并将结果指标存储在Redis中。
事件驱动的缓存刷新：当一个原始的里程碑事件发生时，我们立即使Redis中的所有相关键失效。在2分钟的等待时间后，我们从Neutrino获取所有被废止的键的新结果并更新Redis。2分钟的等待时间是为了确保事件被录入Pinot的实时表。

◆ 性能

通过让货运司机轻松获得他们当前的绩效分数，我们观察到所有关键指标都有了统计学上的显著提升。

迟到的取消 → -0.4
准时提货 → +0.6%。
准时交货 → +1.0%
自动跟踪性能→+1.0%

这些性能的改善，仅在2021年就节省了150万美元的成本。

按业绩评级进行的深入调查显示，被评为 "有风险 "的承运人表现出最大的改善。

这不仅从节约成本的角度显示出很高的商业影响，而且从用户体验的角度来看也是如此。以下是一个推荐书来自Uber货运平台上的一个承运人，她发现这个新功能对她自己的业务有好处

◆ 总结

在这篇博客中，我们描述了Uber货运承运人应用程序中承运人记分卡的后端设计和实现，使用了Apache Pinot和Uber的流媒体基础设施。这种新的架构在生成低延迟（~250ms P99和~50ms P50）的分析指标方面效果惊人，否则就需要在多个在线数据存储上进行复杂的查询。我们的服务已经在生产中可靠地运行了一年多，维护费用非常少。

来源：

https://www.toutiao.com/article/7147027316055507459/?log_from=f353fe227cdbf_1664180912743

“IT大咖说”欢迎广大技术人员投稿，投稿邮箱：aliang@itdks.com