Apache Kafka / Spark流系统的性能调优

电信行业的实际案例研究

调试实际的分布式应用程序可能是一项艰巨的任务。至少在一开始，最常见的Google搜索并没有什么用。在这篇博客文章中，我将详细介绍如何将Apache Kafka / Spark Streaming / Apache Ignite应用程序加速近十倍，并将开发原型转变为有用的，稳定的流媒体应用程序，该应用程序最终超过了性能目标。为应用程序。

此处学习的课程相当笼统，可以轻松地扩展到使用MapR Streams和Kafka的类似系统。

该项目是融合平台需求的具体案例，该平台集成了完整的软件堆栈以支持该系统的需求：实时流，大数据分布式处理和持久性。截至撰写本文时， MapR融合数据平台是此类平台当前唯一可用的生产就绪型实现。

系统目标

为了满足电信公司的需求，该应用程序的目标是将来自三个独立系统的日志数据结合在一起。加入数据后，就可以将网络条件与任何特定客户的特定呼叫相关联，从而使客户支持可以向不满意其电话服务的客户提供准确和有用的信息。如果该应用程序可以实时进行而不是作为批处理工作，则它具有巨大的附加价值，因为6个小时的呼叫质量信息对客户服务或网络运营没有实际价值。

基本上，这是一个相当简单的ETL作业，通常会作为数据仓库的批处理作业完成，但现在必须作为流式分布式体系结构实时进行。

更具体地说，总体情况是将来自远程服务器的输入数据流式传输到分布式集群中，进行一些数据清理和扩充，将三个日志中的记录联接在一起，并将联接的数据作为单个表持久保存到数据库中。

原始系统的问题

原始系统存在几个围绕性能和稳定性的问题。

首先，流应用程序不稳定。在Spark Streaming应用程序中，如果每个微批处理的处理时间等于或小于批处理时间，则称该流稳定。在这种情况下，应用程序的流式传输部分正在30秒的窗口中接收数据，但处理时间为4.5-6分钟。

其次，有一个批处理过程，一次要一个小时连接一次数据，目标是在30分钟内运行，但要花2个小时才能完成。

第三，应用程序运行了几个小时后随机崩溃。

集群硬件，软件堆栈和输入数据

集群硬件非常好，有12个企业服务器节点，每个节点配备两个E5 Xeon CPU，每个CPU具有16个物理核心，256GB内存和八个6TB旋转硬盘。网络是10GB以太网。

为该项目选择的技术栈围绕着Kafka 0.8（用于将数据流式传输到系统中），Apache Spark 1.6（用于ETL操作）（本质上是对输入进行过滤和转换，然后进行联接）以及使用Apache Ignite进行。 1.6作为内存共享缓存，可以很容易地将应用程序的流输入部分与数据连接起来。如果发生故障，Apache Hive还可以用作Ignite的磁盘备份以及单独的分析应用程序。

初始集群的配置如下：

节点	k	神经网络	HDFS	梅索斯	Mesos Master	卡夫卡	火花工人	点燃
1个	X	X	X	X		X	X	X
2	X	X	X	X		X	X	X
3	X		X	X		X	X	X
…			X	X		X	X	X
7			X	X	X		X	X
8			X	X	X		X	X
…			X	X			X	X
12			X	X			X	X

该集群运行Apache Hadoop的HDFS作为分布式存储层，资源由Mesos 0.28管理。最后，HBase用作最终联接数据的最终数据存储。该项目范围之外的其他系统将查询该文件。

系统的性能要求是处理高达3GB / min或150-200,000个事件/秒的输入吞吐量，代表已知的峰值数据吞吐量以及额外的余量。普通吞吐量约为该值的一半，即1.5GB /分钟和60,000-80,000个事件/秒。

原始数据源是三个远程系统的日志，此处标记为A，B和C：日志A占条目的约84-85％，日志B约占1-2％，日志C约占14-15％。数据不平衡这一事实是此应用程序中（许多）困难的来源之一。

Spark应用程序均使用Scala 2.10和Kafka的直接方法（无接收器）进行编码。 Apache Ignite具有一个非常好的Scala API和一个神奇的IgniteRDD ，它可以使应用程序共享内存数据，这是该系统降低编码复杂性的关键功能。

应用架构

单个Kafka生产者将原始数据摄入到运行在6台服务器上的Kafka中。生产者读取各种日志，并将每个日志的记录添加到其自己的主题中。由于有三个日志，因此有三个Kafka主题。每个主题分为36个分区。最有可能存在36个分区，因为有6个节点，每6个磁盘分配给HDFS，Kafka文档似乎建议每个物理磁盘大约有一个分区作为指导。

Spark Streaming应用程序使用数据，该应用程序选择每个主题，然后执行简单的过滤器以切出不必要的字段，进行映射操作以转换数据和执行foreachRDD操作（每个微批处理在Spark Streaming中生成RDD）将数据保存到Ignite和Hive。

流媒体应用程序非常简单：映射，过滤器和foreach分区，保存到Ignite

第二个“常规” Spark应用程序运行在Ignite存储在内存中的数据上，以1小时为批处理将来自三个单独日志的记录合并到一个表中。这项工作是使用Spark的DataFrame API完成的，该API非常适合该任务。第二部分涉及不超过100GB的数据，并且群集硬件的大小适当以处理该数量的数据。

三个小时的数据被累积到Ignite中，因为绝大多数调用持续不到一个小时，并且我们希望一次对一个小时的数据进行连接。由于某些呼叫将在一批中开始而在另一批中完成，因此系统将保留三个小时并且仅处理一个小时的中间批次，因此联接可以成功处理接近100％的记录。

值得一提的是，更好的全流架构可以避免中间表示形式的出现。具有更多时间和事前思考能力的示例性现实案例可以更快地结束整个项目，而不是全神贯注地编写第一个可行的解决方案。

性能调优

这些应用程序的主要问题是由于试图运行开发系统的代码而造成的，这些代码在运行于真实数据的物理，本地群集上的AWS实例上进行了测试。从来没有授予原始开发人员访问生产集群或真实数据的权限。

Apache Ignite是一个巨大的问题源，主要是因为它是一个新项目，没有人有任何实际经验，而且还因为它不是一个非常成熟的项目。

Spark Streaming应用程序在大约4.5分钟内运行，并且项目目标是在30秒内运行。我们需要找到9倍于加速的改进价值，并且由于时间限制，我们无法更改任何代码！

该系统必须在一周内准备好进行生产测试，因此从体系结构和算法角度出发的代码被假定为正确且足够好，以至于我们只有通过调整才能达到性能要求。

修复RPC超时异常

我们从具有相同问题的人那里找到了正确的解决方案，如JIRA的SPARK-14140中所示。他们建议将spark.executor.heartbeatInterval从10s增加到20s。

我认为这个问题可能是由于Kafka，Ignite或垃圾收集器暂停而导致节点由于磁盘繁忙或CPU高峰而变得忙碌所致。由于Spark在所有节点上运行，因此问题是随机的。（请参阅第一部分中的集群服务布局表。）

配置更改完全解决了此问题。从那以后我们再也没有看到过。

增加驱动程序和执行程序的内存

通过将内存从每个执行者20g增加到每个执行者40g以及驱动程序40g，解决了内存不足问题和应用程序的随机崩溃。令人高兴的是，生产集群中的机器配备了大量内存。对于新应用程序，这是一个好习惯，因为您一开始不知道需要多少。

由于Spark UI报告的内存消耗非常小，因此很难精确地调试该问题，缺乏准确的信息。实际上，由于此设置易于更改，因此根据经验，我们将40g作为使应用程序稳定运行的最小内存大小。

增加并行度：增加Kafka中的分区数量

输入数据不平衡，大部分应用程序处理时间都花在处理主题1（吞吐量的85％）上。 Kafka分区与输入RDD中的分区数量进行1：1匹配，导致只有36个分区，这意味着我们只能让36个核心忙于此任务。为了增加并行度，我们需要增加分区数。因此，我们将主题1分为12个主题，每个主题有6个分区，总共72个分区。我们对生产者进行了简单的修改，将第一个日志中的数据平均分为12个主题，而不仅仅是一个。消费者方需要修改零代码。

我们还根据其他两个主题在输入数据中的相对重要性，适当调整了分区数的大小，因此我们将主题2设置为2，将主题3设置为8。

并行运行更多任务。 调整之前，每个阶段始终有36个分区！

调整执行程序的大小

原始应用程序仅运行3个执行程序，共有72个内核。我们将应用程序配置为以80个内核运行，每个执行者最多10个内核，总共8个执行者。请注意，在10个节点的集群中，每个节点具有16个实际核心，我们为Kafka代理，Ignite和HDFS / NN留下了足够的资源。

将批处理窗口从30s增加到1m

生产者每隔30秒将数据分批推送到Kafka，因为它是通过FTP批处理从远程系统收集的。由于需要处理制造商，技术和年龄的困惑范围内的设备和系统，因此这种布置在电信应用中很常见。

这意味着输入流非常不完整，如Spark UI的“流”选项卡的屏幕截图所示：

将窗口增加到1m可使我们平滑输入，并使系统有机会在1分钟或更短的时间内处理数据，但仍保持稳定。

为了确保这一点，该团队生成了一个测试数据，该数据模拟了已知的最坏情况数据，并且使用新设置，火花流工作现在确实很稳定。该团队还能够轻松地在测试数据和实际生产数据流之间进行切换，并通过限制生产者来配置要传入系统的数据量。这对于快速测试各种配置并查看我们是否取得了进展非常有帮助。

删除要求保存到Hive，仅使用Ignite

与项目经理的讨论表明，Hive实际上并不是流应用程序需求的一部分！主要是因为HBase中的数据可以轻松地被分析使用。同样，在此应用程序的上下文中，每个单独的记录实际上都不需要100％保证地进行处理。

确实，根据系统的目标，丢失数据的最坏情况是无法找到客户的呼叫质量信息……情况已经如此。换句话说，数据丢失的风险不是破坏交易的因素，而获得数据的好处是更多的见解。只要处理和存储绝大多数数据，就可以实现业务目标。

所有优化的结果

流媒体应用程序最终稳定下来，优化运行时间为30-35s。

事实证明，淘汰Hive还加快了将数据连接在一起的第二个Spark应用程序的运行，因此它现在的运行时间为3500万，这意味着这两个应用程序现在都符合项目要求。

随着下一部分的改进，Spark Streaming作业的最终性能下降到20s的较低范围，最终加速了12倍以上。

我们必须在稳定性方面下大力气。需要采取几种策略，如下所述。

使Spark Streaming应用程序稳定

我们为修复性能所做的工作直接影响了系统的稳定性。如果两个应用程序本身都稳定并且在适当大小的资源上运行，则系统最有可能总体上保持稳定。

删除Mesos并使用Spark Standalone

Mesos最初选择管理资源是前瞻性的，但最终我们决定将其从最终生产系统中删除。首先，计划是让Mesos管理所有应用程序。但是团队永远无法让Kafka和Ignite与Mesos保持良好的合作关系，因此他们以独立模式运行，仅由Spark由Mesos管理。当然，随着时间的推移，毫无疑问，所有应用程序都可以正确配置为与Mesos一起使用。

提议删除Mesos有点争议，因为Mesos比在独立模式下运行的Spark更先进，更酷。

但是Mesos的问题是双重的：

对执行程序大小和数量的控制很差，这是Spark 1.6的一个已知问题（ SPARK-5095 ），已在Spark 2.0中修复。
Ignite和Kafka不在Mesos内部运行，只是Spark。由于日程安排的压力，该团队已放弃尝试使这两个服务在Mesos中运行。

如果Mesos实际控制资源，则它只能分配好资源。就此系统而言，Kafka和Ignite的运行超出了Mesos的知识范围，这意味着它将错误地将资源分配给Spark应用程序。

此外，它是一个单一用途的集群，因此我们可以使用系统资源的全局视图为每个应用程序自定义资源的大小。几乎不需要动态资源分配，调度队列，多租户和其他流行语。

更改点燃记忆模型

一个已知的问题是，当由JVM控制的堆变得很大（> 32GB）时，垃圾回收的成本会很大。当加入应用程序运行时，我们确实可以看到这个问题：25GB随机播放的阶段中有些行的GC时间峰值很大，从10秒到超过一分钟不等。

Ignite的初始配置是运行ONHEAP_TIERED，并在堆上缓存48GB的数据，然后溢出降至12GB的堆外内存。该设置已更改为OFFHEAP_TIERED模型。尽管由于序列化成本而稍慢，但是OFFHEAP_TIERED不会导致大量垃圾回收。它仍然在内存中运行，因此我们估计这将是净收益。

进行此更改后，每个批次的运行时间从30秒降低到了约25秒，减少了约5秒钟。此外，连续的批处理往往具有更多相似的处理时间，增量为1-3秒，而先前的变化会超过5至10秒。

更新Ignite JVM设置

我们遵循了Ignite文档的性能调整部分（ http://apacheignite.gridgain.org/docs/jvm-and-system-tuning ）中推荐的JVM选项。

完善Spark代码

代码的某些部分假定可靠性，例如对Ignite的查询，而实际上却存在操作失败的可能性。这些问题可以在代码中解决，现在可以更优雅地处理异常，尽管可能还有很多工作可以提高代码的健壮性。我们只能通过立即运行该应用程序来找到这些位置。

将ZooKeeper重新分配给节点10-12

鉴于群集是中型的，因此有必要尽可能多地扩展服务。我们将ZooKeeper服务从节点1-3移到了节点10-12。

结论

调整此应用程序大约需要1周的全职工作。我们使用的主要信息是Spark UI和Spark日志，可以从Spark UI轻松访问。作业和阶段以及流UI的视图确实非常有用。

我学到的是

将流应用程序从AWS上的原型迁移到本地集群需要安排测试时间
不使用真实数据测试AWS原型是一个大错误
包括许多对可靠性要求很高的“出血边缘” OSS组件（Apache Ignite和Mesos）是不现实的
更好的架构设计可以极大地简化系统
调整Kafka / Spark Streaming应用程序需要对整个系统有一个整体的了解。这不仅仅是改变Spark的参数值；它是数据流特征，应用程序目标和对客户的价值，硬件和服务，应用程序代码，然后使用Spark参数的组合。
MapR融合数据平台将减少该项目的开发时间，复杂性和成本。

该项目是这家特定电信公司的第一个项目，他们决定全力开发这种先进的100％开放源代码平台。他们的开拓精神应受到赞扬。但是，更好的平台和应用程序体系结构选择将使他们的生活更加轻松。