Spark Streaming应用程序【部署、升级、监控】

分手后的思念是犯贱 2023-03-01 06:24 6阅读 0赞

## 一.部署应用 ##

要部署运行Spark Streaming应用程序，需要具备以下条件。

*  使用集群管理器进行集群部署，这是任何Spark应用程序的一般要求。
 *  将应用程序JAR打包 ，必须将流式应用程序编译为JAR。如果spark-submit用于启动应用程序，则无需在JAR中提供Spark和Spark Streaming。但是，如果应用程序使用高级数据源【例如Kafka】，则必须将它们链接到的额外工件及其依赖项打包在用于部署应用程序的JAR中。例如，使用的应用程序KafkaUtils 必须将spark-streaming-kafka-x-xx\_x.xx及其所有传递依赖项在其应用程序JAR中包含。
 *  为执行者配置足够的内存，由于必须将接收到的数据存储在内存中，因此必须为执行者配置足够的内存来保存接收到的数据。请注意，如果要执行10分钟的窗口操作，则系统必须在内存中至少保留最近10分钟的数据。因此，应用程序的内存要求取决于其中使用的操作。
 *  配置检查点，如果流应用程序需要它，则必须将Hadoop API兼容的容错存储中的目录（例如HDFS，S3等）配置为检查点目录，并且以这样的方式编写流应用程序，用于故障恢复。
 *  配置应用程序驱动程序的自动重新启动，要从驱动程序故障中自动恢复，用于运行流式应用程序的部署基础结构必须监视驱动程序进程，并在驱动程序失败时重新启动。不同的集群管理器 具有不同的工具来实现这一目标。
    
     *  Spark Standalone，可以提交Spark应用程序驱动程序以在Spark Standalone集群内运行，应用程序驱动程序本身在工作程序节点之一上运行。此外，可以指示独立集群管理器监督驱动程序，并在驱动程序由于非零退出代码或由于运行该驱动程序的节点故障而失败时重新启动它。
     *  YARN，Yarn支持自动重启应用程序的类似机制。
     *  Mesos，Mesos已经实现这一目标。
 *  配置预写日志，自Spark 1.2起，引入了预写日志以实现强大的容错保证。如果启用，则将从接收器接收到的所有数据写入配置检查点目录中的预写日志。这样可以防止驱动程序恢复时丢失数据，从而确保零数据丢失。可以通过设置配置参数 spark.streaming.receiver.writeAheadLog.enable=true来启动。但是，这些更强的语义可能会以单个接收器的接收吞吐量为代价。可以通过并行运行更多接收器来纠正此问题，以增加总吞吐量。此外，建议在启用预写日志时禁用Spark中接收到的数据的复制，因为该日志已存储在复制的存储系统中。可以通过将输入流的存储级别设置为StorageLevel.MEMORY\_AND\_DISK\_SER来完成此操作。在将S3（或任何不支持刷新的文件系统）用于预写日志时，请记住启用 spark.streaming.driver.writeAheadLog.closeFileAfterWrite和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。请注意，启用I/O加密后，Spark不会加密写入预写日志的数据。如果需要对预写日志数据进行加密，则应将其存储在本机支持加密的文件系统中。
 *  设置最大接收速率，如果群集资源不足以使流应用程序能够以最快的速度处理数据，则可以通过设置记录/秒的最大速率限制来限制接收器的速率，接收器对应参数为spark.streaming.receiver.maxRate，Kafka对应参数为spark.streaming.kafka.maxRatePerPartition。在Spark 1.5中，引入了一个称为背压的功能，该功能消除了设置此速率限制的需要，因为Spark Streaming会自动计算出速率限制，并在处理条件发生变化时动态调整它们。这个背压可以通过设置配置参数  
    spark.streaming.backpressure.enabled=true来启用。

## 二.升级应用 ##

如果需要使用新的应用程序代码升级正在运行的Spark Streaming应用程序，则有两种可能的机制。

*  升级后的Spark Streaming应用程序将启动并与现有应用程序并行运行。一旦新的（与旧应用程序接收的数据相同）的数据被预热并准备好，则选择合适的时段将旧的关闭。请注意，对于支持将数据发送到两个目标的数据源（即，较早和升级的应用程序），可以这样做。
 *  现有应用程序正常关闭（StreamingContext.stop(…) 或JavaStreamingContext.stop(…)正常关闭程序），以确保在关闭之前已完全处理已接收的数据。然后可以启动升级的应用程序，它将从较早版本的应用程序退出的同一点开始处理。请注意，只能使用支持源端缓冲的输入源（如Kafka）来完成此操作，因为在上一个应用程序关闭且升级的应用程序尚未启动时，需要缓冲数据。并且无法完成从升级前代码的较早检查点信息重新启动的操作。检查点信息本质上包含序列化的Scala / Java / Python对象，尝试使用经过修改的新类反序列化对象可能会导致错误。在这种情况下，请使用其他检查点目录启动升级的应用程序，或者删除先前的检查点目录。

## 三.监控应用 ##

除了Spark的监视功能外，Spark Streaming还具有其他特定功能。使用StreamingContext时， Spark Web UI会显示一个附加Streaming选项卡，其中显示有关正在运行的接收器（接收器是否处于活动状态，接收到的记录数，接收器错误等）和已完成的批处理（批处理时间，排队延迟等）的统计信息。这可用于监视流应用程序的进度。

Web UI中的以下两个指标特别重要：

*  处理时间，处理每批数据的时间。
 *  调度延迟，批处理在队列中等待先前批处理完成的时间。

如果批处理时间始终大于批处理时间间隔或排队延迟持续增加，则表明系统无法像生成批处理数据一样快的处理完该批数据。在这种情况下，请考虑减少批处理时间。

还可以使用StreamingListener界面监视Spark Streaming程序的进度，该界面可以获取接收器状态和处理时间。请注意，这是一个开发人员API，将来可能会得到改进。