10 月,开发者不可错过的开源大数据大会-2021 WeDataSphere 社区大会深圳站 了解详情
写点什么

Uber 如何为近实时特性构建可伸缩流管道?

2021 年 9 月 23 日

Uber如何为近实时特性构建可伸缩流管道?

背景


Uber 致力于为全球客户提供可靠的服务。要达到这个目标,我们很大程度上依靠机器学习来作出明智的决定,如预测和增益。所以,用来产生机器学习数据和特征的实时流管道已经越来越受到重视。


Uber 公司使用了 Apache Flink 来建立实时流管道,并建立像 Gairos AthenaX 这样的平台来简化开发过程。但是,由于计算的复杂性或需要处理的实时数据量,仍有很多挑战,如扩展性。


本文中,我们将以生产需求和供应特征为例,介绍我们所面临的一些挑战以及如何应对这些挑战。尤其要说明的是,如何使用性能调整框架来优化实时管道。

架构


下图显示了 Apache Flink 中的流管道负责特征计算和提取的架构。我们将在下文详细讨论这些管道。


图 1:简化的架构概述


特征计算


本节详细介绍了如何通过地理空间和时间维度以及全局产品(UberX 等)对任何给定的六边形(参见此处)的原始事件,例如需求和供应事件进行聚合。以下是简化的计算算法:


在一分钟窗口内,按六边形和全局产品类型计算出不同乘客和司机所发生的原始事件数量。在一分钟窗口内,将 Kring Smooth 应用多个环,最多 20 个环(稍后进行讨论)。将每一环的平滑值聚合在多个滑动窗口大小上,最长可达 32 分钟。


总的来说,一条实时管道每分钟为一个六边形生成 54 个特征,使用 9 个环(0,1,2,3,4,5,10,15,20)和 6 个窗口大小(1,2,4,8,16,32)的组合。


接下来,我们讨论算法的第二步。

Kring Smooth


Kring Smooth 过程通过向其 Kring 邻居广播一个六边形的事件计数来计算地理空间聚合。换句话说,某一特定环的六边形的特征值考虑到了该环内所有六边形的事件计数。


为了计算给定六边形 h 在环 r 上聚合的特征值,公式为:


其中, 是环 的六边形数量; 是环 的第 个六边形; 是来自六边形 的事件数。


因此,让我们看看下面的例子,看看如何计算这三个特征的值:六边形 A 的第 0 环、第 1 环和第 2 环,公式如下:


图 2:六边形 A 的第 0 环、第 1 环、第 2 环


该管道按照方程式计算出多个环形尺寸的特征值,最多可达 20。

时间聚合


在一分钟窗口的 Kring Smooth 完成后,算法的第 3 步是将平滑的事件计数在更大的窗口上聚合,最长可达 32 分钟。要计算给定的六边形 H 在更大窗口上的聚集,公式如下:


其中, 是一个窗口的起始时间戳; 是窗口的大小,以分钟为单位; 是来自 Kring Smooth 的平滑事件计数。


下图 3 展示了如何计算 2 分钟窗口的六边形 A 的特征值:


  • 和 数学公式: 窗口的 Kring Smooth 的平滑事件计数分别为 1.0 和 3.0,分别在 分钟和 分钟发出;

  • 2 分钟窗口的特征值为 2.0,通过使用平滑的事件计数,按照上述方程计算,其时间范围为 ( 分钟)。


图 3:六边形 A 的 2 分钟窗口的聚合


流实现与优化


本节以需求管道为例,说明如何在 Apache Kafka 和 Apache Flink 中实现特征计算算法,以及如何调整实时管道。

逻辑作业拓扑


下图 4 说明了计算需求特征的流管道的逻辑 DAG。对于所有尺寸大于 1 分钟的窗口来说,它们是滑动窗口,这些窗口将以 1 分钟为单位滑动,这意味着一个输入事件可能包含在 63 个窗口内:32 + 16 + 8 + 4 + 2 + 1。


图 4:需求管道的逻辑 DAG


下表列出了逻辑 DAG 中主要运算符的功能:

表 1:需求管道的逻辑运算符

流管道的数据量


本节列出了需求管道的数据量:


  • Kafka 主题的平均输入速率:120k/s

  • 六角形的计数:5M

  • 城市的数量:1500

  • 每个城市的六边形平均数和最大数:4000 和 76000

  • 1 分钟内六边形需求事件的平均计数:45

  • 环 20 的六边形计数:1261


显然,该管道具有高容量、密集的计算和大的状态需要管理。第一版实际上是按照逻辑 DAG 构建的,由于包括背压和 OOM 等问题,无法稳定运行(如下图仪表板所示)。由于我们的目标是接近实时的延迟(小于 5 分钟), 因此我们面临的真正挑战是如何建立稳定的工作通道。


内存监视器:


图 5:已用内存的仪表板


延迟监视器:


图 6:延迟的仪表板


如何优化


本节讨论如何调整流管道。Uber 已开发出一种流程管道性能调优框架,并提供端到端集成测试框架。在启动实际调整之前,Uber 就已经开发了专门的集成测试,使我们能够重构或优化流管道,并确信管道仍将产生正确的结果,类似于单元测试,保护我们免受回归。在整个优化过程中,这些集成测试变得非常有价值。


下面,我们将介绍性能调优框架。

性能调优框架


从下面的内三角可以看出,我们的框架集中在三个领域。通过 Uber uMonitor 系统提供的度量标准对网络、 CPU 和内存进行测量和监控。外五边形的顶点表示可以探索主要优化领域。


图 7:性能调优框架


下表简要解释了每个领域的技术和潜在影响:

表 2:性能调优的领域


接下来,我们讨论如何优化管道。

优化


我们对流管道进行了许多优化,一些优化技术对上述多个领域都有影响。其中一项特别的技术:自定义滑动窗口,对所有三个领域都有重大影响,所以我们有一个专门的章节来讨论它,还有一个章节讨论存储。

网络优化


主要的优化技术列于下表:


表 3:网络优化技术


正如上文所详述的,关键的改进是既提供较少的信息,也提供较小的信息。

内存优化


下表列出了内存技术:


表 4:内存优化技术


注:还包括一些用于网络优化的技术。

CPU 优化


应用于 CPU 优化的技术如下:


自定义滑动窗口


仅通过上面的调整,管道仍无法顺利运行,因为它需要数个滑动窗口 (2、4、8、16、32)进行聚合。由于需要按一个键划分事件,窗口聚合的开销如下:


  • 从上游向窗口运算符传递消息时的 De/Ser;

  • 通过网络传输消息;

  • 反序列化时正在创建的对象;

  • 窗口管理所需的状态管理和元数据,如窗口触发器。


这样的开销会对垃圾收集器、CPU 和网络造成巨大压力。更有甚者,滑动窗口比翻滚或固定尺寸的窗口需要更多的状态,因为一个事件需要保存在一系列滑动窗口中。就拿一个 4 分钟的滑动窗口来说:给定一个事件发生在 2021-01-01 T1:15:01 Z,此事件保存在下面的 4 分钟窗口中:


  • 2021-01-01T01:12:00Z ~ 2021-01-01T01:16:00Z

  • 2021-01-01T01:13:00Z ~ 2021-01-01T01:17:00Z

  • 2021-01-01T01:14:00Z ~ 2021-01-01T01:18:00Z

  • 2021-01-01T01:15:00Z ~ 2021-01-01T01:19:00Z


因为滑动窗口的扇出效应,给管道的状态管理带来很大的压力。针对这些问题,我们采用 FlatMap 运算符手工实现了滑动窗口逻辑,其特点如下:


  • 如果允许重用对象,则向上游运算符传递并重用事件,从而避免分区和相关开销。

  • 该状态在内存中被管理,因此每个事件实际上只能复制一份数据。


对于在内存中保存状态所需的最大内存,我们估计如下:


Total Memory = Count(Hexagon)  Count(Product)  Max(window size) * sizeof(event)
= 3M 6 32 * 237b
= 136G
复制代码


对于 128 的并行性,每个容器的内存约为 1G,这是可管理的。在生产中,实际的内存远远低于最大值,因为不是所有的六边形都有时间范围的事件。


这个自定义滑动窗口的效率非常显著,所以我们已经成功地将这个运算符重新用于超过 5 个不同的用例,这些用例需要在多个大型滑动窗口上进行聚合。

优化后的最终作业 DAG


图 8:需求管道的最终 DAG


通过对其进行优化,最终得到了一个更简单的作业 DAG,其中自定义滑动窗口代替了较大的窗口运算符。


如下面的 24 小时仪表板所示,管道始终可靠地运行:


延迟监视器:


图 9:优化后显示延迟的仪表板


容器内存监视器:


图 10:优化后显示内存使用情况的仪表板

存储


为简化管道维护和重新使用 sink,我们对管道 DAG 进行了进一步重构,在 Flink 中将 sink 运算符分离为专门的发布器作业,并将计算和发布器作业与 Kafka 连接起来。此部分主要关注此发布器作业的细节:


而在服务模型中,它会根据地理、时间和产品的要求查询供应信息。我们选择了 Docstore (Uber 的内部 KV 商店解决方案)作为存储。


我们从一个 docstore 集群开始,它由许多用例共享。


下面是每个 API 调用插入一行的结果。写入的 QPS 在 13000 左右达到峰值,但大多数时候都是几百的数量级。


图 11:如果每个 API 调用只有一行,那么编写 QPS 就不稳定

批处理


我们尝试对这些行进行批处理写入,看看能否增加吞吐量。为使批处理更高效,我们基于 Docstore 中的分片号来划分数据。但是,应用批处理后,写入的 QPS 较低。经过深入的研究,我们发现这是因为流作业中所发出的一种度量的一个维度基数过大。我们将这一维改为常数字符串,而非随机的 UUID。写入的 QPS 可以达到 16000 左右。


在写到 Docstore 之前,我们先把数据写到 Kafka 主题。在禁用 Kafka sink 后,我们可以看到写入 QPS 增加了 10% 左右。


在我们把每个分片的批处理量改为 50 后,写的 QPS 增加了一倍,达到 34000。我们还尝试了批处理规模为 100 和 200。对于批处理大小为 100,写 QPS 增加到 37000(大约增加 20%)。


将批处理大小改为 200 后,没有发现有太大的差别。


在下表中,我们列出了不同配置下的 QPS:


表 6:不同批处理大小下的吞吐量

并行性


Flink 作业的并行性是我们为提高 QPS 而调整的另一个参数。


在将发布器作业的并行性更新为 256 后,写入的 QPS 约为 75000,增加了一倍多。批处理小为 200,在并行度为 1024 时,我们看到 QPS 达到 112000。但是,我们发现存在大量的超时错误。将批处理改为 50 后,写 QPS 约为 120000。


表 7:不同作业并行性下的吞吐量

线程池


对于每个 Flink 作业,我们也尝试使用线程池来提高写 QPS,结果如下:


表 8:不同线程池大小下的吞吐量


如果我们使用线程池大小为 16,峰值 QPS 约为 120000,但是这并不太稳定。


经过对共享集群所能想到的所有优化之后,它仍然不能达到写 QPS 的要求。为了进行测试,我们要求一个特殊的集群。

分区调优


移除 Docstoresink,仅保留 FlatMap。没有对分区器的调用,那么 64 个容器就能处理超过 200000 的输入消息率,而不会延迟。


在 FlatMap 之前,我们添加了自定义分区策略。



对于 384 个容器,延迟时间大约是 12 分钟。分区器的延迟范围为 0.2~5 毫秒。当增加到 512 个容器时,延迟降低到 3 分钟。随后,我们发现每个分区器调用的 0.2 毫秒成为瓶颈。在 flatmap 中,我们添加了本地分区器调用缓存。20 分钟后,缓存的点击率类似于输入信息率。


但是,延迟性仍在增加:


图 12:作业延迟现象持续增加。


背压处于自定义分区阶段。


图 13:作业和背压的拓扑处于自定义分区阶段


将并行性更新为 128,有效地消除了管道中的任何延迟性。每个 DC 都可以写入 300000 QPS,没有任何问题。

数据大小


我们尝试了 3 种不同的模式来观察数据大小的差异。第一种模式为每个(环的大小,时间桶,供应/需求)元组使用一个列。第二种模式为需求和供应各使用一张地图。第三种是将颗粒度为 9 级的 7 个六边形分组为一行。



通过 6 天的数据,我们得到的数据大小如下:


表 9:不同数据模式下的压缩


在启用压缩之后,我们可以看到 3 个表可以节省大约 60% 的磁盘。

服务


在测试过程中,我们发现了一些延迟问题。P99 大约有 150 毫秒的延迟。在我们的定价工作流程中,这是不能接受的。经过调试,我们发现每个分区键都有许多行——大约 6000。这就是说,数据库引擎需要扫描至少 6000 行,然后在查询中应用传递的过滤。当分区键大小增加时,就会周期性地出现 200 毫秒的峰值。但我们知道 TTL 也是为这个表设置的,因此我们所做的就是在 Query 中部署一个热补丁,将结果限制在只有未过期的行上,然后应用查询中传递的过滤。这样降低了对底层引擎的扫描,而 P99 延迟降低到 10 毫秒。


图 14:优化之后,服务延迟从 150 毫秒下降到 10 毫秒


结论


考虑到计算逻辑的复杂性、写吞吐量、服务 SLA 等因素,为具有近实时特性的机器学习模型提供性能非常具有挑战性。通过本文,我们介绍了我们所面临的问题和解决方法,希望对类似用例的同行们有所帮助。


作者介绍:


Feng Xu,Uber 高级软件工程师,领导 Gairos、uMetric 的流计算框架。


Gang Zhao,Uber 前高级软件工程师,负责 Gairos 优化、UMetric 消费,同时专注于存储层 Elasticsearch、Apache Pinot、Apache Cassandra、Docstore 和查询层优化。


原文链接:


https://eng.uber.com/building-scalable-streaming-pipelines/

2021 年 9 月 23 日 14:011669

评论

发布
暂无评论
发现更多内容

京东云新一代自研云服务器 4 月上线;COLING 2020丨面向机器阅读理解的双向认知思维网络

京东科技开发者

人工智能 开发者 云服务器

PT100热电阻温度阻值对应表

不脱发的程序猿

数据分析 PT100 28天挑战 3月日更 温度传感器

这操作真香!一线互联网企业高级Android工程师面试题大全,面试真题解析

欢喜学安卓

android 程序员 面试 移动开发

选择IDaaS解决方案的6个技巧

龙归科技

Idaas

深度分析前端构建工具:Vite2 v.s Snowpack3 v.s. Webpack5

智联大前端

vite webpack 构建工具

书单|互联网企业面试案头书之程序员技术篇

博文视点Broadview

区块链电子合同签署平台,区块链电子存证

13530558032

区块链+版权-助力电子微版权保护

13530558032

《谷歌是如何运营的》-读书笔记

曦语

读书笔记

如果延迟退休势在必行,区块链如何助力“养老助老”?

旺链科技

产业区块链

LinqToExcel.Extend 源码分析 第二波

happlyfox

28天写作 3月日更

盘点 HashMap 的实现原理及面试题

老王说编程

Java hashmap HashMap底层原理

USB2.0 扩展器(一拖四)原理图、PCB,可打样使用

不脱发的程序猿

电路设计 USB电路 USB转TTL 28天挑战 3月日更

【LeetCode】删除字符串中的所有相邻重复项Java题解

HQ数字卡

算法 LeetCode 28天写作 3月日更

从产品经理到产品架构师

博文视点Broadview

AI辅助宫颈癌筛查技术全球居首,守护者的力量来源是?

华为云开发者社区

华为云 目标检测 AI技术 宫颈癌

低代码开发平台解决方案之“金融服务行业”篇

优秀

低代码

中国人工智能,赏花更要寻根

脑极体

当跨国企业女职业经理人遇上创业女 CEO,两者会擦出什么样的火花?

科技新消息

如何凝聚党员力量?智慧组工系统构架组织部管理平台解决方案

源中瑞-龙先生

解决方案 党员 智慧组工

面试看这个就够了!最新BAT大厂面试者整理的Android面试题目模板,先收藏了

欢喜学安卓

android 程序员 面试 移动开发

【点评必看】这道 Hard 到底难在哪里?大概是难在考察的是违反“人性直觉”的内容吧 ...

宫水三叶的刷题日记

面试 LeetCode 数据结构与算法

2021金三银四阿里+腾讯+字节+滴滴+美团java面试题全梳理(高清脑图+面试答案)!

程序员小毕

Java spring 程序员 面试 分布式

AI不仅可以把李焕英带回2021,还能告诉你贾玲更像爸爸还是妈妈

京东科技开发者

人工智能 语音识别 语音合成

BFAI量化交易系统开发|BFAI炒币机器人APP软件开发

开發I852946OIIO

系统开发

Java面试热门技术框架:Spring Security Oauth2.0认证授权

Java架构追梦

Java spring 面试 金三银四跳槽

谷歌大佬回国发展,吊打各大厂面试官!吐血总结大厂面试高频点及笔记解析

Java架构之路

Java 程序员 架构 面试 编程语言

JVM笔记 -- JVM的生命周期介绍

秦怀杂货店

JVM 生命周期

能源管理可视化破冰而出,数字孪生打破传统运维僵局

一只数据鲸鱼

物联网 数据可视化 3D可视化 能源管理 智慧电厂

为什么我们开发 San 项目时要用 CLI?

百度Geek说

service SLI san command

2021最新发布拼多多/字节/360/网易/面经总结

比伯

Java 编程 程序员 架构 面试

开源中间件技术学习路线

开源中间件技术学习路线

Uber如何为近实时特性构建可伸缩流管道?-InfoQ