百万TPS高吞吐、秒级低延迟，阿里搜索离线平台如何实现？
导读：阿里主搜（淘宝天猫搜索）是搜索离线平台非常重要的一个业务，具有数据量大、一对多的表很多、源表的总数多和热点数据等特性。对于将主搜这种逻辑复杂的大数据量应用迁移到搜索离线平台总是不缺少性能的挑战，搜索离线平台经过哪些优化最终实现全量高吞吐、增量低延迟的呢？文章大纲如下：
- 前言
- 搜索离线平台基本概念
- 主搜业务特点与性能要求
- Blink Job 性能调优详解
- 结语
本文将与大家分享阿里主搜在实现全量高吞吐、增量低延迟方面的优化经验，希望对大家的 Flink 应用有所帮助。
一.前言
在阿里搜索工程体系中我们把搜索引擎、在线算分等毫秒级响应用户请求的服务称之为“在线”服务；与之相对应的，将各种来源数据转换处理后送入搜索引擎等“在线”服务的系统统称为“离线”系统。搜索离线平台作为搜索引擎的数据提供方，是集团各业务接入搜索的必经之路，也是整个搜索链路上极为重要的一环，离线产出数据的质量和速度直接影响到下游业务的用户体验。
搜索离线平台经过多年沉淀，不仅承载了集团内大量搜索业务，在云上也有不少弹外客户，随着平台功能的丰富，Blink（阿里内部版本的 Flink） 版本的领先，我们在 2019 年年初开始计划把主搜（淘宝天猫搜索）迁移到搜索离线平台上。
主搜在迁移搜索离线平台之前的架构具有架构老化、Blink 版本低、运维困难、计算框架不统一等不少缺点，随着老主搜人员流失以及运维难度与日俱增，重构工作早已迫上眉睫。
对于将主搜这种逻辑复杂的 X 亿数据量级应用迁移到搜索离线平台总是不缺少性能的挑战，业务特点与性能要求决定了主搜上平台的过程中每一步都会很艰辛。为了让性能达到要求，我们几乎对每个 Blink Job 都进行了单独调优，最初的理想与最后的结局都是美好的，但过程却是极其曲折的，本文将主要介绍主搜在迁移搜索离线平台过程中在性能调优方面具体做了哪些尝试。
主搜迁移搜索离线平台的完成对于平台来说有里程碑式的意义，代表搜索离线平台有能力承接超大型业务。
二.搜索离线平台基本概念
搜索离线平台处理一次主搜全增量主要由同步层和数据处理层组成，它们又分别包括全量和增量流程。为了读者更好理解下文，先简单介绍几个关于搜索离线平台的基本概念。
1.集团内支撑业务
目前搜索离线平台在集团内支持了包括主搜，AE 在内的几百个业务。其中数据量最大的为淘宝天猫评价业务，数据量达到了 X 百亿条，每条数据近上 X 个字段。
2.场景
处理用户的数据源（MySQL 或 ODPS）表，将数据经过一系列的离线处理流程，最终导入到 HA3 在线搜索引擎或 ES 中。
3.平台相关技术栈
如下图，搜索离线平台目前数据存储基于 HDFS/盘古，资源调度依赖于 YARN 或 Hippo，计算框架统一用 Flink/Blink 执行。
4.全量
全量是指将搜索业务数据全部重新处理生成，并传送给在线引擎，一般是每天一次。
这么做有两个原因：有业务数据是 Daily 更新；引擎需要全量数据来高效的进行索引整理和预处理，提高在线服务效率。全量主要分为同步层与数据处理层。
5.增量
增量是指将上游数据源实时发生的数据变化更新到在线引擎中。
这也就意味着在我们的场景中对于增量数据不需要保证 Exactly Once 语义，只需要保证 At Least Once 语义。基于该背景，我们才能用全链路异步化的思维来解一对多问题（下文会详细讲解）。
与全量一样，增量也分为同步层与数据处理层。
6.一对多
在搜索这个领域某些业务数据需要用一对多的形式来描述，比如商品宝贝和 SKU 的关系即是个典型的一对多数据的例子。在搜索离线基于 Hologres（阿里巴巴自研分布式数据库）存储的架构中，一对多的数据存储在单独的一张双 pk 的 HoloTable 中，第一、二主键分别的宝贝 ID 与 SKU_ID。
有了上面这些概念之后，在后续的段落中我们会看到搜索离线平台针对主搜各 Blink Job 的性能调优，先简要概括下主搜业务特点与性能要求。
7.数据存储方式
搜索离线平台以前用 HBase 做镜像表时，是用一张多列族大宽表来存储业务单维度所有数据。经过详细调研之后，我们决定用 Hologres 替换 HBase，所以需要对存储架构做全面的重构。用多表来模拟 HBase 中的多列族，单 HoloTable 中包括很多业务数据源表的数据。重构后的数据存储方式大致如下：
8.同步层
所谓同步层，一般是将上游数据源的数据同步到镜像表，供数据处理层高效处理。由于业务方单维度的数据有很多 MySQL 表或 ODPS 表组成，少则 X 张，多则像主搜这样 X 张。所以将同纬度数据聚合到一张 Holo 表中时，如果多张表两两 join 的话会产生大量 shuffle，所以我们采取异步 upsert 方式，不同数据源表的数据写 Holo 表中不同的列来解决海量数据导入问题。
9.数据处理层
所谓数据处理层，是指将同步层得到的各镜像表（HBase/Holo）的数据进行计算，一般包括多表 Join、UDTF 等，以方便搜索业务的开发和接入。
三.主搜业务特点与性能要求
下面首先介绍下主搜业务特点与性能要求，再详细介绍我们进行了怎样的调优才达到了性能的要求。
1.主搜业务特点
- 数据量大
主搜有 X 亿（有效的 X 亿）个商品，也就是主维度有 X 亿条数据，相比于平台其他业务（除淘宝评价业务）多出 X 个数量级。这么多数据我们能否在 X 个多小时完成全量？如何实现高吞吐？挑战非常大。
- 一对多的表很多
主搜业务有很多一对多的表需要 Join，例如一个商品对应多个 SKU，部分商品对应了接近 X 个 SKU 信息。这些信息如何能够高性能的转换为商品维度，并与商品信息关联？
- 源表的总数多
主搜有 X 多张表（包括一对多的表），平台其他业务的源表个数一般都在个位数。源表数量多会导致一系列的问题，比如读取 ODPS 数据时如何避免触发 ODPS 的限制？拉取大表数据时如何做到高吞吐？这些问题都需要我们一一解决。
- 热点数据
主搜有一些大卖家（饿了么，盒马等）对应了很多商品，导致在数据处理层出现非常严重的数据倾斜等问题。如何解决大数据处理方向经常出现的 SKEW？
2.主搜性能要求
- 全量（同步层 + 数据处理层）高吞吐！
全量要求每天一次，在有限的资源情况下每次处理 X 亿的商品，这么大的数据量，如何实现高吞吐，挑战非常大！
- 增量（同步层 + 数据处理层）低延迟！
增量要在 TPS 为 X W 的情况下达到秒级低延迟，并且双 11 期间有部分表（例如 XX 表）的 TPS 能达到 X W，增量如何保证稳定的低延迟？值得思考！
下面一一描述我们是如何解决这些问题来达到性能要求的。
四.Blink Job 性能调优详解
根据上述主搜业务特点与性能要求罗列出下图，左边与中间两列表示主搜哪些特点导致某阶段任务性能差。所以我们要对相应阶段 Blink Job 进行调优，调优完成也就代表着平台能满足图中最右边一列主搜所需要的全量高吞吐与增量低延迟的性能要求。
下面按照全量，增量，解一对多问题的脉络来给大家介绍我们是如何解决上述五个问题之后达到全量高吞吐以及增量低延迟的性能要求的。
1.全量高吞吐性能调优
全量主要包括同步层与数据处理层，必须实现高吞吐才能让全量在 X 个多小时之内完成。同步层在短时间内要同步约X张表中的上 X 亿全量数据，且不影响同时在运行的增量时效性是一个巨大的挑战。数据处理层要在短时间内处理X多亿条数据，Join 很多张镜像表，以及 UDTF 处理，MultiGet 等，最后产生全量 HDFS 文件，优化过程一度让人频临放弃。这里重点介绍数据处理层的性能调优历程。
该 Job 的调优历时较长，尝试方案较多，下面按照时间顺序讲解。
- 初始形态
首先提一下 IC 维度为商品维度，UIC 维度为卖家维度，并且最开始我们的方案是没有 FullDynamicNestedAggregation 和 IncDynamicNestedAggregation 的（后文会详细提到这两个Job）。Scan IC 维度单 Pk 表之后做一系列的 DImJoin、UDTF、MultiJoin。在测试过程中发现 DimJoin 多 pk 表（一对多表）的数据时，性能非常低下，全链路 Async 的流程退化成了 Sync，原因是我们一对多的数据存在单独的一个 SaroTable（对多个 HoloTable 的逻辑抽象）中，对指定第一 pk 来取对应所有数据用的是 Partial Scan，这是完全 Sync 的，每 Get 一次都要创建一个 Scanner，虽然我们不但对于 DimJoin 加了 Cache，并且对于主搜特有的 MultiGet 也加了对于 SubKey 的精准 Cache。但是测试下来发现，性能还是完全得不到满足，所以尝试继续优化。
- 引入 LocalJoin 与 SortMergeJoin
由于性能瓶颈是在 DimJoin 多 pk 的 SaroTable 这里，所以我们想办法把这部分去掉。由于一对多的 SaroTable 只有两个维度具有，所以我们尝试先分别将 IC 维度与 UIC 维度的所有表（包括单 pk 与多 pk）进行 LocalJoin，结果再进行 SortMergeJoin，然后继续别的流程。
首先介绍下 Local Join。由于 HoloStore 保证相同 DB 中所有表都是按照相同的 Partition 策略，并且都是按照主键字典序排好序的，所以我们可以将同纬度同 Partition 的数据拉取到一个进程中进行 Join，避免了 Shuffle，如下图所示。
所以拓扑大概变为：
经过测试，由于业务上面存在大卖家（一个卖家有很多商品），导致 SortMergeJoin 之后会有很严重的长尾，如下图所示，Uid 为 101 与 103 的数据都是落到同一个并发中，我曾经尝试再这个基础之上再加一层 PartitionBy nid 打散，发现无济于事，因为 SortMergeJoin 的 Sort 阶段以及 External Shuflle 对于大数据量的 Task 需要多次进行 Disk File Merge，所以该长尾 Task 还是需要很长时间才能 Finish。
- 加盐打散大卖家
所以我们需要继续调优。经过组内讨论我们决定对大卖家进行加盐打散，从 ODPS 源表中找出 Top X 的大卖家 ID，然后分别在主辅维度 Scan + Local Join 之后分别加上 UDF 与 UDTF，具体流程图与原理示例见下面两幅图：
如上图所示，Uid 为 101 与 103 的数据被打散到多个并发中了，并且因为我们在 SortMergeJoin 之后加了 UDTF 把加的 Salt 去掉，所以最终数据不会有任何影响。
- 最终形态
这样全量 FullJoin 总算完成了，并且性能也勉强达标，所以我们开始调整增量流程（IncJoin），这时发现 IncJoin 跟 FullJoin 的初始形态存在一样的问题，追增量非常慢，永远追不上，所以组内讨论之后决定在同步层针对全量新增一个 FullDynamicNestedAggregation Job（下文会详细提到），这是一个 Blink Batch Job 它将各维度一对多的 SaroTable 数据写到对应维度的主表中，然后在 FullJoin 最开始 Scan 时一起 Scan 出来，这样就避免了 DimJoin 多 pk 的 SaroTable。最终达到了全量高吞吐的要求，全量 FullJoin 最终形态如下：
2.增量低延迟性能调优
增量性能主要受困于数据处理层 IncJoin，该 Job 最开始是一个 Blink Stream Job，主要是从 SwiftQueue 中读出增量消息再关联各个镜像表中的数据来补全字段，以及对数据进行 UDTF 处理等，最后将增量消息发往在线引擎 SwiftQueue 中。
基于“流批一体”的思想，经过一系列尝试，我们增量数据处理层 Job 的最终形态如下。
与全量不同的是由于增量是实时更新的，所以更新记录不仅要写到 Swift Queue 中，还要写入 SaroTable 中。另外，我们根据业务特点给各个 Job 分别加了按 pk 对记录去重的 window。
3.解一对多问题
主搜有很多一对多的表，在数据处理层如何高效的将数据 Get 出来转换为主维度之后进行字段补全，困扰我们很久。
为了提升效率我们必须想办法提升 Cpu 利用率。所以 Get 记录改为全链路异步来实现，由于我们一对多数据存在多 pk 的 HoloTable 中，指定第一 pk 去获取相关数据在 Holo 服务端是以 Scan 来实现的。这样由于异步编程的传染性，全链路异步会退化为同步，性能完全不达标。
- 解决方法
为了将“伪异步”变成真正的全链路异步，经过多次讨论与实践之后，我们决定将一对多表中相同第一 pk 的多条数据 Scan 出来 GroupBy 为一条数据，将每个字段转化为 Json 之后再 Put 进主表中，主要步骤如下图所示。
我们针对全量与增量在同步层加 Job 来解决，分别为 FullDynamicNestedAggregation（Blink Batch Job）与 IncDynamicNestedAggregation（Blink Stream Job），这两个 Job 大致流程为如下图所示。
值得一提的是，正如前文介绍增量时提到的背景，我们的场景中对于增量数据不需要保证 Exactly Once 语义，只需要保证 At Least Once 语义。所以基于该背景，我们能够将数据处理层增量 Job 拆分为两个 Job 执行，一对多的问题得以解决。
这样我们在数据处理层就不需要去 Scan HoloTable 了，从而可以用全链路异步化的方式来提升增量整体性能。
- 截断优化
为了避免将多条数据转为一条数据之后由于数据量过大导致 FullGC 的“大行”问题。基于业务的特性，我们对于每个一对多表在 Scan 时支持截断功能，对于相同的第一 pk 记录，只 Scan 一定条数的记录出来组装为 Json，并且可以针对不同的表实现白名单配置。
- 加过滤 Window 优化
针对业务的特点，一对多的很多表虽然可以接受一定时间的延迟，但是为了避免对离线系统以及在线 BuildService 造成太大的冲击，所以更新不能太多，所以我们加了 30min 的去重窗口，这个窗口作用非常大，平均去重率高达 X% 以上。
五.结语
经过一系列优化，主搜不仅在资源上相对于老架构有不少的节省，而且同时实现全量高吞吐与增量低延迟，并且在 2019 年度双 11 零点应对突增流量时表现的游刃有余。
对系统进行性能调优是极其复杂且较精细的工作，非常具有技术挑战性。不仅需要对所选用技术工具（Flink/Blink）熟悉，而且对于业务也必须了解。加 window，截断优化，加盐打散大卖家等正是因为业务场景能容忍这些方法所带来的相应缺点才能做的。
除了本文提到的调优经验，我们对同步层全增量 Job 与 MultiGet 也进行了不少调优，篇幅原因与二八原则这里就不详细介绍了。
主搜成功迁移也使得搜索离线平台完成了最后一块拼图，成为阿里巴巴集团搜索中台以及核心链路的基础模块。
作者简介：
王伟骏，花名鸿历，阿里巴巴搜索推荐事业部高级开发工程师。2016 年硕士毕业于南京邮电大学。Apache Hadoop & Flink & Eagle Contributor。目前负责阿里巴巴搜索离线平台 Runtime 层相关工作。
另外，陈华曦（昆仑）给了本文很多建议，文中部分图由李国鼎（石及）贡献。
