Apache Spark Streaming,如何处理下游依赖性故障

Question

我试图了解如何使Spark Streaming应用程序更容错(特别是在尝试写入下游依赖项时),并且我不知道在尝试将结果写入外部源时处理失败的最佳方法是什么,像Cassandra,DynamoDB等.

例如,我有一个Spark Streaming作业从Stream(Kafka,Flume等等)中提取数据......我还没有最终确定使用哪种技术,将类似项目聚合在一起,然后将结果写入外部存储.(即Cassandra,DynamoDB或任何正在接收我的DStream计算结果的东西).

我试图弄清楚如何处理外部依赖关系无法写入的情况.也许集群发生故障,可能存在权限问题等,但我的工作无法将结果写入外部依赖项.有没有办法暂停Spark Streaming以便接收器不会继续批量处理数据？我应该只是睡觉当前批次并让接收器继续存储批次吗？如果问题是暂时的(几秒钟),继续批处理可能是可以接受的,但如果依赖性下降几分钟或1小时以上会发生什么？

我有一个想法是有一个监视进程,在后台监视依赖项的健康状况,如果它发现它"不健康",它将停止工作.然后,当所有依赖项都运行正常时,我可以重新启动作业并处理未写入外部源的所有数据.

我的另一个想法是以某种方式在DStream forEachRdd方法中发出信号,表示存在问题.我可以在DStream中抛出某种异常,它会向驱动程序发出它应该停止的信号吗？

如果有人有任何关于如何处理外部容错的经验,或者可以指向我的好文章/视频,那将是很好的.

谢谢

Answer 1

我相信这里没有简单且普遍的答案。很大程度上取决于应用程序语义、数据源类型（可靠接收器、可靠接收器、基于文件、无接收器）和要求。

一般来说，您不应该让应用程序对单个 IO 故障进行故障转移。假设你有一些行动：

outputAction[T](rdd: RDD[T]): Unit = ???

至少确保它不会向您的驱动程序传播异常。

outputActionWithDelay[T](d: Duration)(rdd: RDD[T]) = ???

stream foreachRDD { rdd => Try(outputAction(rdd)) }

问题仍然是下一步做什么。您可以做的最简单的事情就是删除给定的窗口。根据应用程序，它可能是可接受的解决方案，也可能不是可接受的解决方案，但一般来说，在许多情况下丢失一些数据是完全可以接受的。

可以通过跟踪故障并在达到某个阈值时采取其他操作来进一步改进。

如果删除数据不可接受，下一步是在延迟一段时间后重试：

outputActionWithDelay[T](d: Duration)(rdd: RDD[T]) = ???

stream foreachRDD { 
  rdd => Try(outputAction(rdd))
    .recoverWith { case _ => Try(outputActionWithDelay(d1)(rdd)) }
    .recoverWith { case _ => Try(outputActionWithDelay(d2)(rdd)) }
   ...
}

重试次数和延迟持续时间因情况而异，并且取决于存储传入数据的来源和能力。

当我们最后一次重试时，您可以做什么？对于初学者，我们可以添加替代输出源。例如，您可以将所有内容推送到可靠的外部文件存储，然后再担心，而不是使用主要来源。如果输出源需要输入数据的特定顺序，则这可能不适用，但在其他情况下应该值得尝试。

alternativeOutputAction[T](rdd: RDD[T]) = ???

stream foreachRDD { 
  rdd => Try(outputAction(rdd))
    .recoverWith { case _ => Try(outputActionWithDelay(d1)
    ...
    .recoverWith { case _ => Try(outputActionWithDelay(dn)(rdd)) }
    .recoverWith { case _ => Try(alternativeOutputAction(rdd))
}

如果失败，则可能是严重问题的征兆，并且我们在应用程序级别无能为力。我们可以回到第一种方法，只是希望情况能尽快解决，或者选择更复杂的方法。

如果输入源可以缓冲数据并且我们使用可靠的存储和复制，那么我们可以启用检查点并简单地终止应用程序。

如果您尝试恢复，添加断路器的某些变体可能是个好主意，并且如果应用程序在尝试到达主输出时遇到多次失败，则立即丢弃恢复尝试。