如何使用 Spark 3.0 中新加的 Structured Streaming UI 来进行异常分析

本文为阿里巴巴技术专家余根茂在社区发的一篇文章。php

Structured Streaming 最初是在 Apache Spark 2.0 中引入的，它已被证实是构建分布式流处理应用程序的最佳平台。SQL/Dataset/DataFrame API 和 Spark 的内置函数的统一使得开发人员能够轻松实现复杂的需求，好比支持流聚合、流-流 Join 和窗口。自从 Structured Streaming 发布以来，社区的开发人员常常要求须要更好的方法来管理他们的流做业，就像咱们在 Spark Streaming 中所作的那样。为此，Apache Spark 3.0 为 Structured Streaming 开发了一套全新的 UI。html

新的 Structured Streaming UI 经过有用的信息和统计信息提供了一种简单的方法来监控全部流做业，从而使开发调试期间的故障排除变得更容易，在生产环境下经过实时度量更好的理解咱们的做业瓶颈。新的 UI 提供了两组统计信息：java

流查询做业的聚合信息；
流查询的详细统计信息，包括输入速率（Input Rate）、处理速率（Process Rate）、输入行数（Input Rows）、批处理持续时间（Batch Duration,）、操做持续时间等（Operation Duration）。

文章目录bootstrap

1 流查询做业的聚合信息
2 详细统计信息
3 使用新的 UI 进行故障排除
4 将来工做

流查询做业的聚合信息

当开发人员提交一个流 SQL 查询时，这个做业的信息将在 Structured Streaming 选项卡中显示，其中包括活动的流查询和已完成的流查询。流查询的一些基本信息将在结果表中列出，包括查询名称、状态、ID、运行 ID、提交时间、查询持续时间、最后一个批次的 ID 以及聚合信息，如平均输入速率和平均处理速率。流查询的状态有三种：运行（RUNNING），完成（FINISHED）以及失败（FAILED）。全部完成的和失败的查询都在已完成的流查询列表中显示。表格中的错误列（Error）显示失败查询的异常详细信息。具体以下：服务器

A look at the new Structured Streaming UI in Apache Spark™ 3.0

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

咱们能够经过单击表格中 Run ID 那列连接查看流查询的详细统计信息。微信

详细统计信息

统计信息页面显示了包括输入/处理速率、延迟和详细操做持续时间在内的指标，这对于洞察流查询的状态很是有用，使咱们可以轻松地调试流做业运行过程当中的异常状况。页面以下所示：dom

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

上图包含如下的监控信息：分布式

Input Rate: 全部数据源数据流入的聚合以后速度
Process Rate: Spark 处理全部数据源的处理速度，也是聚合后的结果。
Batch Duration: 每一个批次处理时间。
Operation Duration: 执行各类操做所花费的时间，以毫秒为单位。

使用新的 UI 进行故障排除

在这一小节，让咱们来看看如何使用 Structured Streaming 新的 UI 来进行异常排除。咱们的测试代码以下：函数

import java.util.UUID

val bootstrapServers = ...
val topics = ...
val checkpointLocation = "/tmp/temporary-" + UUID.randomUUID.toString

val lines = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", bootstrapServers)
    .option("subscribe", topics)
    .load()
    .selectExpr("CAST(value AS STRING)")
    .as[String]

val wordCounts = lines.flatMap(_.split(" ")).groupBy("value").count()

val query = wordCounts.writeStream
    .outputMode("complete")
    .format("console")
    .option("checkpointLocation", checkpointLocation)
    .start()

因为处理能力不足而致使延迟增长

在第一种状况下，咱们运行查询来尽快处理 Apache Kafka 中读取的数据。在每批中，流做业将处理 Kafka 中全部可用的数据。若是咱们的资源不足以快速处理当前批次的数据，那么延迟将迅速增长。最直观的判断是输入行和批处理持续时间会呈线性增加。处理速率（Process Rate）提示流做业最多只能处理大约8,000条记录/秒。可是当前的输入速率大约是每秒 20,000 条记录。咱们能够为流做业提供更多的执行资源，或者添加足够的分区来处理这些数据。oop

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

处理时间比较稳定但延迟很高

这种状况相比第一种状况是处理延迟没有持续增长，具体以下所示：

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

咱们发如今相同的输入速率（Input Rate）下，处理速率（Process Rate）能够保持稳定。这意味着做业的处理能力足以处理输入数据。然而，每批处理的进程持续时间（即延迟）仍然高达20秒。高延迟的主要缘由是每一个批处理中有太多数据须要处理。一般咱们能够经过增长做业的并行性来减小延迟。在为 Spark 任务添加了10个Kafka分区和10个核心以后，咱们发现延迟大约为5秒——比20秒要好得多。

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

使用 Operation Duration 图来进行异常诊断

操做持续时间（Operation Duration）图以毫秒为单位显示执行各类操做所花费的时间。这对于了解每一个批次的时间分布并简化故障排除颇有用。让咱们以Apache Spark 社区中的性能改进 SPARK-30915 为例进行说明。

在 SPARK-30915 工做以前，当压缩后的元数据日志变得很大时，压缩后的下一批处理要比其余批处理花费更多的时间。

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

通过对代码进行分析以后，发现并修复了没必要要的读取压缩日志文件的问题，也就是 SPARK-30915 解决的，下图的运行时间确认了咱们预期的效果：

若是想及时了解Spark、Hadoop或者HBase相关的文章，欢迎关注微信公众号： iteblog_hadoop

将来工做

经过上面三个案例，新的 Structured Streaming UI 将帮助开发人员经过更加有用的流查询信息来更好地监视流做业。做为早期发布版本，新的 UI 仍在开发中，并将在之后的版本中进行改进，包括但不限于如下功能：

更多流查询执行细节：延迟数据（late data），水印（watermark），状态数据指标（state data metrics）等等。
Spark 历史服务器中支持 Structured Streaming UI。
针对异常状况的更明显的提示：好比延迟发生等。

本文翻译自：A look at the new Structured Streaming UI in Apache Spark™ 3.0

本博客文章除特别声明，所有都是原创！
转载本文请加上：转载自过往记忆（https://www.iteblog.com/）
本文连接: 【如何使用 Spark 3.0 中新加的 Structured Streaming UI 来进行异常分析】（https://www.iteblog.com/archives/9844.html）