scala - Spark-检查点对性能的影响

Question

从Spark的DStreamCheckpointData看来，检查点机制似乎收集了要检查点的time窗口并将其更新/写入检查点文件。我正在尝试特别了解几件事:

在每个检查点间隔，它是否都读取所有先前的检查点数据，然后更新当前状态？如果是这样，那么当检查点状态变得非常大时，会对性能产生什么影响，这肯定会减慢长时间运行的流上下文的速度。

是否有通用规则或公式来计算不同数据摄取率，滑动窗口和批处理间隔的检查点间隔？

Answer 1

是，检查点是一项阻止操作，因此它在事件期间将停止处理。状态串行化停止计算的时间长短取决于您要写入哪个媒体的写入性能(您是否听说过Tachyon/Alluxio？)。

另一方面，不会在每个新的检查点操作上读取先前的检查点数据:当对流进行操作时，状态信息已经在Spark的缓存中进行了维护(检查点只是其备份)。让我们想象一下最简单的状态，即所有整数的总和，在整数流中满足:在每个批次中，您都将基于在批次中看到的数据为此总和计算一个新值-并可以存储此部分总和在缓存中(请参见上文)。每五批左右(取决于检查点间隔)，您将此总和写入磁盘。现在，如果在随后的批次中丢失了一个执行器(一个分区)，则可以通过仅对该执行器的分区重新处理最多五个分区(通过读取磁盘以查找最后一个检查点)来重建总数。 ，并重新处理最后五个批次的缺失部分)。但是在正常处理(无事件)中，您无需访问磁盘。

我不知道任何通用公式，因为您必须确定要从中恢复的最大数据量。旧文档给出a rule of thumb。

但是对于流来说，可以将批处理间隔视为计算预算。假设您的批处理间隔为30秒。在每个批处理上，您有30秒的时间分配给写入磁盘或计算(批处理时间)。为了确保您的工作稳定，必须确保批处理时间不会超出预算，否则您将填满集群的内存(如果需要35秒来处理和“刷新” 30秒的数据，在每批处理中，您摄取的数据要多于同时刷新的数据-由于您的内存有限，因此最终会产生溢出现象。

假设您的平均批处理时间为25秒。因此，在每个批次中，您的预算中有5秒的未分配时间。您可以将其用于检查点。现在考虑检查点需要花费多长时间(您可以从Spark UI中取笑)。 10秒？ 30秒 ？等一下 ？

如果在c秒批处理时间间隔内花费bi秒到检查点，并具有bp秒批处理时间，您将在以下方面从检查点“恢复”(处理在没有处理的时间内仍存在的数据):
ceil(c / (bi - bp))批次。

如果需要k批处理从检查点“恢复”(即恢复从检查点引起的延迟)，并且您要对每个p批处理进行检查，则需要确保强制执行k < p，以避免作业不稳定。因此，在我们的示例中:

，因此，如果要花10秒才能完成检查点，那么将需要10/(30-25)= 2个批次进行恢复，因此您可以每2个批次(或更多，即不那么频繁)进行检查点，我建议您考虑一下计划外的时间损失)。

，因此，如果花30秒才能完成检查点，则将需要30/(30-25)= 6个批次来恢复，因此您可以每6个批次(或更多)检查一次。

如果需要60秒才能完成检查点，则可以每12个批次(或更多)检查一次。

请注意，这假设您的检查点时间是恒定的，或者至少可以受最大常数的限制。令人遗憾的是，通常情况并非如此:一个常见的错误是忘记使用updateStatebyKey或mapWithState之类的操作删除有状态流中的部分状态-但是状态的大小应始终是有界的。请注意，在 Multi-Tenancy 群集上，写入磁盘的时间并不总是恒定的-其他作业可能正在尝试在同一执行程序上同时访问磁盘，使您无法使用磁盘iops(在this talk中，Cloudera报告在IO吞吐量显着降低之后> 5个并发写入线程)。

请注意，您应该设置检查点间隔，因为默认值是在最后一批之后出现的第一批超过default checkpoint interval(即10s)的批次。对于我们的30s批次间隔示例，这意味着您每隔一个批次都要检查一次。即使纯粹出于容错性的原因(即使重新处理几批也没有那么大的成本)，这也太频繁了，即使您的计算预算允许这样做，也会导致性能图中的以下峰值: