Delhivery 公司最佳实践剖析：从 Apache Kafka 迁移至 Amazon MSK

Original URL: https://thinkwithwp.com/cn/blogs/big-data/best-practices-from-delhivery-on-migrating-from-apache-kafka-to-amazon-msk/

本文为Delhivery公司的客座文章。

在本文中，我们将了解Delhivery公司将运行在Amazon Elastic Compute Cloud (Amazon EC2)上的自建Apache Kafka迁移至Amazon Managed Streaming for Apache Kafka (Amazon MSK)的整个流程与相关操作步骤。Delhivery公司高级技术架构师Akash Deep Verma表示，“这套新架构已经投入生产一年有余，我们在Amazon MSK上运行着超过350款应用程序，它们每一天、每一秒都在产生并消费数据。总体而言，与Amazon MSK合作给我们带来了愉悦的体验！”

Delhivery公司是印度领先的数字商务执行平台。凭借其遍布2500座城市、18000多个行政管辖区域的全国网络，该公司提供一整套涵盖快递包裹运输、短途/长途货运、反向物流、跨境、B2B与B2C仓储以及技术服务的全套物流服务体系。

Verma指出，“我们的愿景是通过世界一流的基础设施、最高质量的物流运营以及尖端工程技术能力的结合，成为印度商务经济的「操作系统」。我们的团队已经成功为印度超过1.2亿个家庭完成了6.5亿份订单。我们运营有24处自动分类中心、75处执行中心、70处物流枢纽、超过2500个直接配送中心、超过8000处合作伙伴中心以及14000多辆货运车辆。超过40000名团队成员每年365年、每周7天、每天24小时为客户服务，正是这一切，让日均百万个包裹的交付配送成为可能。”

自托管Apache Kafka的难题

我们每天需要处理近1 TB数据，借此实现各类分析功能。这些数据来自货运跟踪、订单跟踪、GPS、生物识别、手持设备、分类器、重量、客户以及设施等各类来源，并通过多条实时与批处理管理在各类系统与服务之间往来移动。数据在经过处理与充实之后，即可服务于多种业务及技术场景。考虑到我们业务的基本性质，Apache Kafka上的传入消息与事件主要以稳定常规流量与间歇峰值流量组成，这两类情况的流量区间分别为每秒10000到12000条消息、以及每秒50000到55000条消息。

Apache Kafka则作为这些依赖系统与服务之间的核心消息与事件主干。

我们之前在 Amazon Elastic Compute Cloud (Amazon EC2)实例上对Apache Kafka代理及其相关组件（例如Apache Zookeeper）进行自建。

随着业务规模的不断增长，管理这些组件并保证正常运行时间已经成为一项重要的资源密集操作。我们需要专门指派两名开发人员管理我们的Apache Kafka基础设施并维持其正常运作。由于这些开发人员无法继续为业务功能的开发做出有效贡献，这种无差别的繁重工作已经在事实上导致生产力下降。

Verma表示，“我们想要一项托管形式的Apache Kafka服务，借以降低基础设施管理所占用的时间与资源。这样一来，我们就能重新分配技术团队的事务优先级，保证各位成员专注于开发能够实际增加业务价值的功能。”

迁移至Amazon MSK以节约时间

我们研究了几种选项，考量借此替换EC2实例上的自托管Apache Kafka，最终的答案正是Amazon MSK。借助Amazon MSK，我们可以继续使用原生Apache Kafka API，并在不更改任何代码的前提下在AWS上运行我们的现有应用程序。它还能为我们提供、配置并维护Apache Kafka集群与Apache Zookeeper节点。以此为基础，我们能够将开发人员从基础设施管理当中解放出来，为我们的业务编写更多创新应用。

根据AWS团队的建议，我们采取了以下步骤：

调整MSK集群大小。
将各Apache Kafka主题迁移至Amazon MSK。
监控Amazon MSK运行状态。

调整MSK集群大小

为了适当调整MSK集群的大小，我们需要了解当前工作负载的基本动态。我们从当前基于Amazon EC2的Apache Kafka集群中检索出以下指标：

生产者的写入效率——我们考虑监控broker上的指标BytesInPerSec，选择各代理的平均值，并汇总了集群中全部代理的值以估算净摄取率（请与ReplicationBytesInPerSec指标区分开来，ReplicationBytesInPerSec代表的是与其他代理间的摄取率）。
消费者的消费速率——我们还在broker上监控指标BytesOutPerSec，选择单一代理的平均值，而后将集群中所有代理的值进行汇总，借此估算集群的净消费率（请与ReplicationBytesOutPerSec指标区分开来，ReplicationBytesOutPerSec代表的是指向其他代理的消费率。
数据复制策略——我们通过集群全局参数default.replication.factor以及为每个主题设定的独立参数之间进行了评估，并借此确定了复制因子的最大值。
数据保留策略与磁盘利用率的目标百分比——我们通过集群全局参数log.retention.hours 以及为每个主题设定的独立参数之间进行了评估确定了数据保留的最高保留值。我们还指定了已用存储空间的百分比，并估算满足用例所需要的净空间。

AWS为我们提供一份Amazon MSK规模调整与费率标准表格，帮助我们快速估算MSK集群中所需要的代理数量。接下来，我们在业务环境当中进行概念验证，并从电子表格中找到了推荐的具体集群大小。这份电子表格还帮助我们轻松估算了集群使用成本。关于更多详细信息，请参阅每集群代理数量。

将各Apache Kafka主题迁移至Amazon MSK

我们整理了将各主题迁移至Amazon MSK的几种可行选择：

MirrorMaker 1.0，这是Apache Kafka中随附的一款独立工具，能够以最低停机时间将数据从自托管Apache Kafka集群迁移至Amazon MSK。
使用消费者从自托管Apache Kafka集群读取数据，而后将数据写入至Amazon MSK。这种迁移方法需要一定的停机时间。
我们以往经常使用过的其他复制工具。

在实践操作中，我们将前两种方法结合起来进行迁移。对于无法容忍长时间停机的主题，我们使用MirroMaker 1.0将数据迁移至Amazon MSK。对于能够承受一定停机的主题，我们的内部SLA允许通过应用程序流量将数据从自托管Apache Kafka集群重新定向至Amazon MSK。

在MirrorMaker方案中，我们需要在自托管EC2实例上设置MirrorMaker 1.0守护程序以使用来自源集群的消息，而后将其重新发布至目标MSK集群。每个MirrorMaker 1.0线程都配备一个单独的消费程序实例，且共享同一通用生产程序。整个流程如下：

MirrorMaker 1.0实例产生一个消费者进程，该进程与Apache Zookeeper整体交互，以支持源Apache Kafka集群进行主题发现。
消费者进程从相关主题中读取消息。
MirrorMaker 1.0生成一个生产者进程，该进程通过Apache Zookeeper端点与Amazon MSK上托管的Apache Zookeeper fleet进行交互。
该生产程序进程通过代理端点将消费程序进程检索到的消息，转发至Amazon MSK的相应主题。

下图所示，为我们的整个迁移拓扑结构。

MSK集群创建流程需要将子网ID作为输入，以便代理与Pache ZooKeeper节点能够正确映射至客户VPC。通过在具有主专用IPv4地址的各子网ID中创建ENI，我们能够轻松实现这一映射。MSK集群中随附的代理与Apache ZooKeeper端点将实际解析这些专用IPv4地址。

我们使用以下命令，将所有主题从基于Amazon EC2的源集群以镜像形式保存至Amazon MSK：

kafka-mirror-maker.sh
--consumer.config config/mirrormaker-consumer.properties
--producer.config config/mirrormaker-producer.properties
--whitelist '*'

上述命令包含以下细节：

其中的kafka-mirror-maker.sh shell脚本用于创建一个 tools.MirrorMaker类实例。
在 mirrormaker-consumer.properties 文件中的bootstrap.servers 、group.id等由换行符分隔的键/值对下，包含有消费者配置参数。
其中mirrormaker-producer.propertiesfile在bootstrap.servers、acks等由换行符分隔的键值对下，包含有生产者配置参数。
--whitelist选项允许大家使用Java样式的正则表达式，保证我们可以仅对特定主题进行镜像复制，例如使用--whitelist 'A|B'镜像复制A与B主题。

随着KIP-382的引入，现在Amazon MSK已经能够支持MirroMaker 2.0并发挥由新版本带来的更多优势。关于使用MirroMaker 2.0将自托管Apache Kafka集群迁移至MSK集群的操作说明、配置文件、示例代码以及相关实验，请参阅 MirrorMaker2 on Amazon EC2研讨资料。

大家也可以使用Amazon Kinesis Data Analytics（一项面向Apache Flink的全托管服务）将现有Apache Kafka集群迁移至Amazon MSK。如此一来，您可以使用全托管Apache Flink应用程序处理Amazon MSK中存储的流式数据。关于将Amazon Kinesis Data Analytics与Amazon MSK配合使用的更多详细信息，请参阅教程：使用Kinesis Data Analytics应用程序在同一VPC内实现不同MSK集群间的数据复制以及点击流实验室。

在稳定状态下，我们生产环境中的MSK集群使用以下配置：

broker节点——6 x m5.4xlarge
复制因子——3
生产者数量——110个以上
消费者数量——300个以上
主题——500个以上
broker上运行的Kstreams——55个以上

监控Amazon MSK运行状态

Amazon MSK以三种粒度级别提供多项Amazon CloudWatch监控指标：

DEFAULT
PER_BROKER
PER_TOPIC_PER_BROKER

在我们的用例中，我们使用最精细的PER_TOPIC_PER_BROKER粒度以实现最佳系统运行可见性。为了自动检测各类运营问题，我们还使用以下指标开发出自定义CloudWatch警报。

指标名称	警报触发条件	具体含义
`UnderReplicatedPartitions`	当不等于0时	代理的未完全复制分区的数目。如果分区副本的数据更新情况远远落后于来源，则从ISR池中删除该分区。这时 `IsrShrinksPerSec`指标将对应增加。
`OfflinePartitionsCount`	当不等于0时	集群中处于脱机状态的分区的总数。任何无活动主节点的分区都将无法访问，且在主节点恢复可用之前，对应的消费者与生产者将持续处于阻塞状态。
`IsrShrinksPerSec`	当本身值增加，但 `IsrExpandsPerSec`未正确增加时	副本可能出于以下几个原因而被从ISR池中删除：副本与主节点的偏移量过大。副本在一定时段内未与主节点联系（具体时长可通过`replica.socket.timeout.ms`进行配置）。集群中加入更多代理(`IsrExpandsPerSec` 也应同步增加)。分区被移除(`IsrExpandsPerSec` 也应同步增加)。
`UncleanLeaderElectionsPerSec`	当不等于0时	这里的unclean主节点选举是一类特殊情况，原因是同步中没有可用的副本。由于各个主题必须对应一个主节点，因此如果只能在未同步的各副本中进行主节点选举，则前主节点停机之前未经同步的所有消息都将永远丢失。
`ActiveControllerCount`	当不等于1时	Apache Kafka集群中的控制器负责维护分区主节点列表并协调主节点过渡（在当前主节点不可用时）。
`KafkaDataLogsDiskUsed`	设置一项警报，当磁盘利用率超过85%时，则通过API或者使用控制台添加存储容量，保证磁盘利用率降至70%或更低。	每个broker所对应的EBS存储卷的空间利用百分比。

Amazon MSK还支持通过端口11001（面向JMX Exporter）与端口11002（面向Node Exporter）使用Prometheus等监控工具与Prometheus格式的指标（例如Datadog、New Relic以及SumoLogic等）捕捉公开指标。关于更多详细信息，请参阅使用Prometheus实现公开监控。关于配置公开监控的更多操作说明，请参阅Open Monitoring实验室。

总结

在使用自建Apache Kafka代理时，我们需要保证Apache ZooKeeper始终发挥仲裁作用、监控代理之间的网络连接，同时监控LogCleaner等不同Apache Kafka辅助进程。一旦某个要素发生故障，我们还需要通过基础设施管理与监控等方式解决问题。这部分被耗费在Apache Kafka运营层面的精力与时间，本应被用于更好地实现实际业务价值。

Amazon MSK能够降低基础设施的维护强度，简化问题的识别与解决，缩短代理维护时间，最终将生产力提升至新的层面。它在后台承担起Apache Kafka的维护工作，结合实际需求为我们提供监控级别选择，让我们的团队能够腾出更多精力改善业务应用程序并为客户提供价值回报。

亚马逊AWS官方博客