这是来自阿里巴巴软件工程师于根茂（Genmao Yu）的客座社区帖子。结构化流最初是在apachespark2.0中引入的。它已经被证明是构建分布式流处理应用程序的最佳平台。SQL/Dataset/DataFrame API和Spark的内置函数的统一使开发人员能够轻松地实现其复杂的需求，例如流聚合、流流连接和窗口化支持。自从结构化流媒体发布以来，开发人员经常要求更好的方式来管理他们的流媒体，就像我们在Spark Streaming（即DStream）中所做的那样。在apachespark3.0中，我们为结构化流媒体发布了一个新的可视化UI。新的结构化流式UI提供了一种简单的方法，可以使用有用的信息和统计信息来监视所有流作业，从而更容易在开发调试期间进行故障排除，并通过实时指标提高生产的可观察性。用户界面提供两组统计信息：1）流式查询作业的汇总信息；2）流式查询的详细统计信息，包括输入速率、处理速率、输入行数、批处理持续时间、操作持续时间等。流式查询作业的聚合信息当开发人员提交一个流式SQL查询时，它将被列在Structured streaming选项卡中，其中包括活动流式查询和已完成的流式查询。结果表中会列出流式查询的一些基本信息，包括查询名称、状态、ID、运行ID、提交时间、查询持续时间、最后一批ID以及平均输入率、平均处理率等汇总信息。流式查询的状态有三种，即正在运行、已完成和失败。所有已完成和失败的查询都列在已完成的流式查询表中。错误列显示失败查询的异常详细信息。我们可以通过单击runid链接来查看流式查询的详细统计信息。详细统计信息统计信息页面显示包括输入/处理速率、延迟和详细操作持续时间在内的指标，这些指标对于深入了解流式查询的状态非常有用，使您能够轻松调试查询处理中的异常情况。它包含以下指标：输入速率：数据到达的总速率（跨所有源）。处理速率：Spark处理数据的聚合速率（跨所有源）。批次持续时间：每个批次的持续时间。操作持续时间：执行各种操作所用的时间（毫秒）。跟踪的操作如下：addBatch：从源读取微批的输入数据、对其进行处理并将批处理的输出写入接收器所用的时间。这将占用微批次的大部分时间。getBatch：准备逻辑查询以从源读取当前微批的输入所用的时间。getOffset：查询源是否有新的输入数据所用的时间。walCommit：将偏移量写入元数据日志。queryPlanning：生成执行计划。需要注意的是，并非所有列出的操作都将显示在UI中。在不同类型的数据源上有不同的操作，因此所列操作的一部分可以在一个流式查询中执行。使用UI排除流式处理性能问题在本节中，让我们看几个新的结构化流式UI表示发生了一些不寻常的事情。在较高的层次上，demo查询如下所示，在每种情况下，我们将假设一些先决条件：进口java.util.UUIDval bootstrapServers=。。。val主题=。。。"/val-"/tmp+临时位置UUID.randomUUID.toStringval线=火花.读流.format（"卡夫卡"）.选项（"kafka.bootstrap.servers"，bootstrapServer）.option（"订阅"，主题）.加载（）.selectExpr（"CAST（值为字符串）"）.as[字符串]val字数=线条.平面图（\.split（""）.groupBy（"value"）.count（）val查询=字数.writeStream.outputMode（"完成"）.format（"控制台"）.option（"检查点位置"，检查点位置）。开始（）处理能力不足导致延迟增加在第一次查询Apache的情况下，我们要尽快运行Kafka进程中的数据。在每个批处理中，流作业将处理Kafka中的所有可用数据。如果处理能力不足以处理批处理数据，则延迟将迅速增加。最直观的判断是输入行和批处理持续时间会呈线性上升。处理速率提示流作业最多只能处理8000条记录/秒。但目前的输入速率约为20000条记录/秒。我们可以给流作业更多的执行资源，或者添加足够的分区来处理与生产者保持同步所需的所有消费者。稳定但高延迟对于本节的案例，与前一个有什么不同？延迟不是持续增加，而是保持稳定，如下图所示：我们发现在相同的输入速率下，处理速率可以保持稳定。这意味着作业的处理能力足以处理输入数据。然而，每批处理的持续时间，即延迟，仍然高达20秒。高延迟的主要原因是每个批处理中的数据太多。通常我们可以通过增加这个作业的并行性来减少延迟。在为Spark任务增加10个Kafka分区和10个核心之后，我们发现延迟大约为5秒，比20秒好得多。使用操作持续时间图表进行故障排除"操作持续时间"图表以毫秒为单位显示执行各种操作所用的时间。它有助于了解每个批次的时间分布，并使故障排除更加容易。让我们以apachespark社区中的性能改进"SPARK-30915：查找最新批处理ID时避免读取元数据日志文件"为例。在这项工作之前，当压缩后的元数据日志变得巨大时，压缩后的下一个批比其他批花费更多的时间。在代码调查之后，发现并修复了对压缩日志文件的不必要的读取。下面的操作持续时间图表证实了我们预期的效果：未来发展如上所示，新的结构化流媒体UI将帮助开发人员使用更有用的流式查询信息更好地监视他们的流作业。作为早期版本，新的UI仍在开发中，并将在将来的版本中进行改进。未来还可以实现以下功能，但不限于：更多流式查询执行细节：延迟数据、水印、状态数据度量等。支持Spark history server中的结构化流式UI。针对异常情况的更明显的提示：延迟发生等。尝试新的用户界面在新的Databricks运行时7.1中，在apachespark3.0中试用这个新的Spark流式UI。如果您使用的是Databricks笔记本，它还为您提供了一种查看笔记本中任何流式查询状态和管理查询的简单方法。你可以在Databricks上注册一个免费帐户，几分钟后就可以开始免费使用了，不需要信用卡。 O'Reilly学习火花书免费第二版包括Spark 3.0的更新，包括Pandas UDF的新Python类型提示、新的日期/时间实现等。免费下载