Kafka是一个非常流行的分布式消息队列，它可以实时地支持海量的数据，可以支持非常高的吞吐率和低延时，它在Uber得到了非常广泛的应用，很多的核心业务都会用到它，比如打车服务和送餐服务。

本文将介绍Uber如何搭建一个基于Kafka的跨数据中心复制平台，除了有关Kafka在Uber的应用，还有Uber的Kafka Pipeline，以及在哪些应用场景下会用到跨数据中心的复制等等。

本文整理自徐宏亮在 ArchSummit 全球架构师峰会 2019 北京站的演讲。

今天我要讲的题目是：Uber如何搭建一个基于Kafka的跨数据中心复制平台。

首先我会介绍一下Kafka在Uber的应用；然后是Uber的Kafka Pipeline，以及在哪些应用场景下会用到跨数据中心的复制；之后会介绍我们开发的跨数据中心复制平台uReplicator，最后是我们在复制过程中所做的数据丢失检测。

Apache Kafka at Uber

Kafka是一个非常流行的分布式消息队列，它可以实时地支持海量的数据，可以支持非常高的吞吐率和低延时，它在Uber得到了非常广泛的应用，很多的核心业务都会用到它，比如打车服务和送餐服务。

在使用打车服务时，司机和用户的应用都会将自己当前的信息实时发送到Kafka，我们在Kafka的下游有一个流处理的平台，它从Kafka实时读取数据进行计算，得出动态打车的价格。

而送餐服务有一个重要特性，需要计算出实时的预计送达时间，大家打开应用的时候，会看到每一个餐馆后面都会有一个预计的送达时间，我们的后台会有一个机器学习的平台，实时从Kafka读取商家、司机和用户三方的数据，用机器学习的方法去动态计算这个送达时间。当然还有很多这样的例子，包括我们对诈骗的实时检测、司机和用户的注册，这些都可以基于Kafka。

总结来看，Kafka在Uber的应用主要有以下五大类场景：

第一类是将Kafka作为最基本的消息队列。各种服务和应用都可以将数据发送到Kafka，也可以从Kafka进行读取，不允许服务之间进行直接的通信，从而避免网状的通信；

第二类是将Kafka作为流处理平台数据的来源。在Uber我们有一个开源的流处理平台AthenaX，在这个平台上我们可以运行Samza和Flink的job，它们从Kafka读取数据进行实时计算，计算完成后也可以将数据发回给Kafka，再传回到下游；

第三类是将Kafka作为数据库。我们的数据库会将自己的变更日志通过Kafka传输到下游，也会通过Kafka进行跨数据中心的传输；

第四类是将Kafka作为批量加载的服务。如果想把数据加载到HDFS和S3，首先需要将数据发送到Kafka，再由Kafka统一加载到HDFS和S3；

第五类是将Kafka作为常见服务和应用的日志。所有的服务和应用都可以使用我们提供的库将日志发送到Kafka，然后用户既可以直接从Kafka进行读取来进行调试，也可以通过Kafka将日志加载到Elasticsearch之类的检索平台进行搜索。

在这五类场景中，Kafka在Uber扮演了一个数据中枢的角色，基本上所有的服务都会将数据发送到Kafka，而后再进行读取和进一步的操作。

Apache Kafka pipeline & replication

因为对Kafka有着非常广泛的应用，所以我们的Kafka集群规模也非常大。我们的集群现在有数万个Topic，每天都会处理超过万亿条的数据，每天的数据量也会达到PB的级别。下面我们来看一下Uber构建了怎样的Kafka pipeline，以支持这些应用的场景。

在Uber我们有多个数据中心，每个数据中心都会有一个单独的Kafka pipeline，在每个pipeline里会有多个Regional Kafka集群，每个集群处理不同类型的数据，比如有的集群处理服务日志相关的数据，有的集群处理核心业务相关的数据，从而做到较好地隔离不同类型的数据。

在每一个Kafka的集群前面，我们都会有一个Kafka REST Proxy集群，它其实就是一个REST service，所有应用和服务都会使用所提供的REST Client将数据首先发送到REST Proxy，再由REST Proxy统一发送到Kafka，这样就做到了Client和Kafka broker之间的隔离。

每一个Pipeline最右边都有一个Aggregate Kafka集群，它的作用是收集所有数据中心里Regional Kafka的数据，相当于它会拥有全局的数据。如果一些服务处理需要全局数据的话，可以从Aggregate Kafka读取，再进行进一步的操作。在Regional Kafka和Aggregate Kafka集群之间，由uReplicator跨数据中心、数据平台进行实时复制。

下面，我们来看一下，我们在哪些场景下会需要用到跨数据中心的复制。

第一个应用场景是Aggregation，很多服务都需要一个Global view来处理数据，需要一个复制平台将所有的数据从所有数据中心的Regional集群复制到Aggregate集群。

第二个应用场景，对于那些需要Global view的服务来说，他们通常只运行一个数据中心，当这个数据中心出现问题以后需要切换到另外的数据中心，在切换的过程中，他们需要知道切换之后需要从哪一个offset继续开始读取数据，所以复制平台在复制的过程中，还需要做一个额外操作，实时将复制数据的source和Destination的offset的对应关系发送到一个叫offset Sync的服务，这个服务可以实时收集所有Partition offset的对应关系，计算出从一个数据中心切换到另一个数据中心后需要从哪个offset进行读取。这样既可以保证没有数据丢失，还可以保证将重复数据降到最少。

第三个应用场景是批量加载服务，我们需要首先将所有的数据复制到Aggregate集群，再统一加载到HDFS或S3，这样可以避免零星加载，也可以使加载过程做到更好的批量化。

最后一个应用场景是，我们可以利用这个复制平台来做Topic的迁移，既可以从一个数据中心迁移到另外一个数据中心，也可以在一个数据中心里将一个集群迁移到另外一个集群。Topic迁移的难点在于通常Producer和Consumer都不在一个组，也有可能会有多个Producer和Consumer，所以我们很难做到让它们在同一个时间进行切换。

我们的做法是，从新的数据中心往老的数据中心迁移时，先搭建一个复制的集群进行数据的复制。我们可以把Producer首先迁移到新的数据中心，因为有了这个复制集群，所以两个数据中心会有相同的数据，这样Consumer就可以在Producer迁移完之后，再慢慢迁移。如果我们有多个Consumer，他们也可以按照自己的计划分别进行迁移，当所有的Consumer都迁移完以后，我们才可以把这个复制集群移除，至此迁移的过程就算完成了。这么做的好处是，在迁移的过程中可以把对Consumer的down time降到最低。

uReplicator

那么，我们为什么想要开发一个我们自己的跨数据中心复制平台呢？
2015年，我们使用的是Kafka自带的MirrorMaker，在使用的过程中碰到了很多的问题。

首先，它使用了Kafka的High-Level Consumer，所以就不可避免地会发生昂贵的、非预期的Consumer Rebalance，虽然Rebalance可以更好地将Partition分配到Worker上，但High-Level Consumer Rebalance有几个缺点。

第一，不可以控制它什么时候发生，比如增加或减少Topic时可能会触发Rebalance，增加或减少Worker时也可能会触发Rebalance；

第二， Rebalance的过程中所有复制都会停止，这段时间任何数据都无法流入到下流，这非常不好。另外，在Rebalance结束之后，会留下这段时间的backlog，MirrorMaker会有一个巨大的traffic去Catch-up这些backlog，这样的spark对Destination的broker也有非常大的影响。

第三，当时MirrorMaker使用了一个静态的Topic List，需要在MirrorMaker集群启动的时候读取，想要增加或者减少一个Topic时，需要更新这个静态的文件列表，再重启整个MirrorMaker集群。每次都要维护静态文件很不方便，而且在重启集群的时候又会导致触发Rebalance的问题。

第四，当时的MirrorMaker无法控制Consumer Commit offset的时间，经常出现把数据发送到了Kafka的producer里，但是还没有实际送到Destination的情况。这个时候，Consumer就Commit了offset，在这种情况下，如果这一个MirrorMaker的Worker Crash了，恢复以后就不会再去重复复制这些信息，那么这些信息就彻底丢失了。

最后，因为MirrorMaker使用了一些静态的配置文件，当这些文件没有被同步到所有Worker上的机器时，启动时就会发生各种问题。

所以基于以上这些问题，我们决定开发一个自己的复制平台。我们对这个平台提出了五个要求：

Stable replication

希望这个平台有稳定复制的能力，如果有Topic或者Worker在复制过程发生变动，复制行为不应该受到影响，还可以继续稳定复制。

Simple operations

希望可以方便地增加或者删除Topic，而不需要维护静态文件，不需要在每次改变Topic的时候重启整个集群。

High throughput

希望这个平台有非常好的性能，以满足Uber数据量的增长。

No data loss

希望在复制的过程中可以保证没有数据丢失。

Auditing

和上一个相关，希望有机制可以检测是否真的发生了数据丢失。

基于这些要求，我们开发了第一代uReplicator，引入了分布式的任务管理框架Apache Helix。简单来说，它由Controller、Participant、Resource几部分构成。在我们uReplicator的具体应用中，Controller就相当于我们的uReplicator Controller，可以用来做Partition的分配；Participant相当于实际用来进行复制操作的每个机器；而Resource就是每个Kafka Topic。

Helix的工作原理是在Controller里通过自带的算法或者用户自定义的算法进行一定的计算，把每个Resource以Partition为单位，分配到Worker上，Worker在收到以后再进行进一步的操作。所以，在引入Helix以后，我们相当于取代了High-Level Consumer里面的Consumer Group Coordinato角色，完全自己控制Partition分配的行为。这样我们直接解决了两个问题，第一是稳定复制的能力，当Topic或者Worker发生变动的时候，可以控制这个变动所带来的影响；第二个好处是，我们在Controller里加了一个End Point，这样我们就可以使用curl命令，在运行的时候动态地进行Topic的增加和删除，这样就弃掉了静态的Topic List，不需要再重启整个集群。

我们还对MirrorMaker的Worker行了一系列的改进，把它演变成了我们的uReplicator的Worker。我们没有改变模式，但将MirrorMaker Worker里面High-Level Consumer替换成了SimpleConsumer，这样我们就有了更多底层的控制，可以控制更多Consumer的行为，并可以更好发控制Simple Consumer的数量，达到更高的并行度，增加吞吐量。另外我们修改了Commit Offset的逻辑，可以保证在运行的过程中，只有当Worker在Producer里的所有数据都被发送到Destination broker，并且收到了发送成功的返回以后再去Commit Offset，从而保证数据不会被丢失。如果Worker crash了，它就可以从上一个Commit Offset进行复制，虽然会有数据重复，但可以避免数据丢失。

Performance Issues

在大概2017年初的时候，我们发现随着Uber数据量的增长，性能已经不能满足当时Uber的吞吐量。当时我们做了很多研究，提出了很多解决方案。在此会介绍其中的4个解决方案。

Solution 1: Increase Batch Size

首先我们在改进的过程中发现，在Catch-up的过程中分为两个阶段，一个是全速（Full-speed）阶段，另外一个是Long tail阶段。全速阶段的定义是在这个过程中，所有的Worker上都含有非常高的Backlog，每个Worker都在全速Catch-up这些Backlog。我们在这个阶段观察到的第一个现象，Destination broker上的CPU利用率非常高，从Kafka的指标来看，它上面RequestHandler的利用率也非常高，这说明每个broker都收到了太多发送的请求，由于CPU的限制，没办法把每一个发送的请求更快地处理掉，接收更多发送的请求。

对此问题一个最直观的解决方案是，增大每一个发送请求的Batch Size，从而减少请求的频率或者数目，我们相应地增加了uReplicator Worker里一些Kafka Producer的参数，比如batch size和linger time，增加这些参数之后效果非常明显，每个request batch size增大了将近四到五倍，同时因为batch size变大，压缩比例也变得更好了。

Solution 2: 1-1 Partition Mapping

我们发现Kafka Producer默认会用Round robin的方式，将数据发送到Destination的每个Partition，但这样带来一个问题，假设我们有一个Topic，它有N个Partition，通常会被分配到N个Worker上，每一个Worker都会往Destination的N个Partition上发送数据，所以一个Topic就相当于产生了N方个连接。可以想象一下，我们有数万个Topic，当所有连接都发生的时候，相当于我们的复制平台对我们自己的Kafka的集群进行了一个DDOS攻击。

我们对于这个问题的解决方案是引入了一个1-1 Partition Mapping，如果一个Message在Source是Partition I，我们把它发送到Destination时，也会继续把它发送到Partition I，这样我们所有连接的数量就和Partition的数量成正比，很大地减轻了Destination Broker上的压力。在采用了这些改进措施以后，我们每个Destination Broker的吞吐量提升了将近1倍，并且这个吞吐量是压缩比更高的吞吐量。

Solution 3: Dynamic Workload Balance

下面我们来看一下，对Long Tail Phase做了哪些改进。 Long Tail Phase的定义是一些Worker已经Catch-up了backlog，变得比较空闲，但另外一些Worker它还在全速进行Catch-up。

一个非常直观的解决方案是，我们希望将那些还在忙碌的Worker上的Workload分配到那些已经空闲的Worker上，使它们都可以全速地工作。所以我们提出的第一个解决方案是动态Workload balance，一开始uReplicator、MirrorMaker进行Partition分配的逻辑，是希望每个Worker都有相同数量的Partition，以求平均。这么做本身并没有问题，但是当数据量特别大的情况下，一些Workload比较大的Partition可能就会被分配到同一个Worker上，导致那个Worker被overload，就不能Catch-up所有的Backlog。

我们的做法不是用Partition的数量来进行分配，而是用Workload来进行分配，在我们的内部有一个叫Chaperone的服务，它可以实时提供每一个Topic在Source Kafka集群里面的bytes-in-rate，我们的Controller就会从这个服务去实时获取Workload，根据Workload进行partition的分配，最终做到每个Worker上都有相同数量的Workload。

Solution 4: Lag-Feedback Rebalance

我们发现在数据中心切换的场景下，很多时候一些Workload并不大的Partition也可能有比较大的backlog，如果只按照前面说的Workload去做Balance，就没有办法为它们分配到更多的资源，但其实在Lag非常大的情况下，即使这些Topic的Load不大，每个Worker也都需要分配出相当一部分带宽去进行复制。

所以我们引入了一个Lag-Feedback Rebalance机制，我们认为在Lag非常大的情况下，它相对应的Workload也变大了。所以我们在计算Workload的时候引入了一个系数，这个系数会随着Lag变大，从而将Lag和整体的Workload统一。

我们采用了动态Workload Rebalance后，每隔十分钟会做一次Rebalance，每做一次Rebalance之后，每个Worker上Partition的数量都会有一些细微的调整，Workload或者说Lag特别大的一些Worker，它上面有很大的Backlog，在Rebalance之后它的Backlog就被挪到了一些空闲的Worker上。从而使所有的Worker都可以全速地Catch-up。

至此，我们已经满足了一开始提出的前四项要求，该去实现最后一项要求了。但很不幸的是，在2017年年终的时候，我们发现前两个要求又得不到满足了，这是由于多方面的原因。

第一，随着Uber数据量的增长，我们每一个Kafka集群的规模也在不断增大，每个集群里Partition的数量也在增长。在当时，我们uReplicator的部署方式是，一个Source到Destination的Kafka的集群由一个uReplicator部署来负责，对于最大的那个Kafka集群来说，那个uReplicator里面需要负责处理非常多的Partition。所以Helix在这个时候就碰到了一个计算上的瓶颈，我们发现在做rolling restart的时候，可能需要将近一个小时才能做一次full Balance，在这一个小时里，很多Partition就停止了复制，于是稳定复制的要求就不能得到满足了。

第二，在当时Uber Kafka集群的数量也在不断的增长，Kafka部署的数量通常会跟Kafka集群的数量成倍数关系，一旦我们又想要有新的Kafka集群的时候，需要去建立更多uReplicator的部署。当我们有这么大的部署量时，就很难去维护它们了，很难控制每个部署里面运行的版本号。

第三，因为有那么多的uReplicator部署，我们很容易就会引入一些人为产生的错误，比如有一次，我们whitelist一个Topic，不小心在两个Kafka集群之间相互whitelist，就变成了一个死循环。那个Topic本身也就10QPS，最后因为相互的复制，变成了100KQPS。另外一个可能出现的人为错误是，从一个Source到Destination可能有不同的路径，可能有时候会错误地将它们在两个不同的路径上对同一个Topic进行了whitelist，这样就会重复复制同一份数据。

基于以上这些问题，我们开发了第二代uReplicator，我们叫它Federated uReplicator，它是在第一代uReplicator的基础上，加入了一个Federation层。

在Federation模式下，第一个解决的是Partition数量过多的问题。我们的做法是将Partition数量非常大的每一个uReplicator部署拆分成几个更小的部署，叫做route。我们在每个route里规定了Partition数目的上限，有了这个上限，使每个route里面的Helix计算Partition分配时变得非常快。

第二个需要解决的是部署数量过多的问题。在第一代uReplicator里，每个部署都需要自己或者用Script人为完成，我们为此引入了一个叫做Deployment Group的概念，它的思想是把相同类型的部署聚合到一个Group里。打个比方，可能我们原来有几十个都是用来做跟Aggregation相关的部署，现在我们把它聚合到一个Aggregation Deployment Group，原来我们可能需要人为地去直接面对几十个Deployment，现在，我们就只需要面对一个Deployment Group。这样不管有多少Kafka集群，我们只需要面对有限个类型的Deployment Group。

在每一个Deployment Group里，都会有一个叫uReplicator Manager的模块来进行每个route的创建和维护工作。Manager的第一个工作是收到一个Topic whitelist请求的时候，会决定是把这个Topic加入到已有的route里面，还是重新建一个新的route。它创建新route的条件是需要保证现有的每一个route里Partition的数目不会超过我们规定的上限。Manager的第二个工作是，它也会从chaperone服务获取route里所有Topic的Workload，然后计算整个route有多少Workload，据此计算出这个route需要多少个Worker进行复制可以达到一定的吞吐率。它同时可以阶段性的去获取数据，动态进行Worker的调整，当Workload变少以后就可以自动释放这个route里的一些Worker，可以更好地利用硬件资源，不造成浪费。

在uReplicator Manager之前有一个叫做uReplicator Front的模块，使用户可以直接做whitelist/blacklist Topic。在这个模块里有一个功能，会将whitelist的信息都存在数据库里，当它收到一个whitelist的请求时候，会从数据库或者内存里将这个Topic相关的whitelist信息都读取出来，我们可以把每一个whitelist想象成一个有向图，在收到whitelist请求以后，它可以判断在这个有向图里新加入一条边后，会不会造成Loop和Double route的情况。当所有检查都通过后，它才会将这个whitelist请求发送到Manage，去做实际whitelist的操作。

Data loss detection

在我们开发了Federated uReplicator之后，终于满足了一开始提出的前四项要求。现在介绍一下，我们是怎么做数据丢失检测的。我们做数据丢失检测的服务，就是刚才提到的chaperone服务。它不光可以提供Workload信息，还可以做数据丢失的检测。

它的具体做法是，在整个Kafka pipeline的每个tier上分别Kafka message数量的进行统计，汇总到Cassandra，然后比较Regional Kafka和Aggregate Kafka中Message的数量，从而判断是否有数据的丢失。

最终，我们满足了一开始在开发这个复制平台时提出的所有要求，uReplicator和Chaperone都在我们的生产环境里运行了两年以上，这两个都是我们的开源项目，在Github上可以找到它们的源代码。

创作场景

Uber 如何搭建一个基于 Kafka 的跨数据中心复制平台