使用 Dataflow 托管式服务执行流水线时，您可以通过 Dataflow 的 Web 监控界面查看该作业和其他任何作业。通过此监控界面，您可以查看相关 Dataflow 作业并与之交互。
您可以使用 Google Cloud Console 访问 Dataflow 监控界面。此监控界面可以显示以下内容：
- 包含当前正在运行的所有 Dataflow 作业和过去 30 天内运行的作业的列表。
- 每个流水线的图形表示。
- 有关作业状态、执行情况和 SDK 版本的详情。
- 运行流水线的 Google Cloud 服务（如 Compute Engine 和 Cloud Storage）相关信息的链接。
- 作业执行期间出现的任何错误或警告。
您可以在 Dataflow 监控界面中查看作业监控图表。这些图表显示流水线作业持续时间内的各种指标，并包含以下信息：
- 步骤级可见性，可帮您确定哪些步骤会导致流水线延迟问题。
- 可揭示出异常行为的统计信息。
- 可帮助确定来源和接收器中的瓶颈问题的 I/O 指标。
访问 Dataflow 监控界面
如需访问 Dataflow 监控界面，请按以下步骤操作：
- 登录 Cloud Console。
- 选择您的 Google Cloud 项目。
- 点击左上角的菜单。
- 转到大数据部分并点击 Dataflow。
此时系统将显示 Dataflow 作业及其状态的列表。如果您没有看到任何作业，则需要运行新作业。如需了解如何运行作业，请参阅 Dataflow 快速入门。
作业可能具有以下状态：
- —：监控界面尚未从 Dataflow 服务收到状态。
- 正在运行：作业目前正在运行。
- 正在启动...：作业已创建，但系统在启动前需要一些时间进行准备。
- 已加入队列：一个 FlexRS 作业已加入队列。
- 正在取消…：正在取消作业。
- 已取消：用户已取消作业。
- 正在排空…：正在排空作业。
- 已排空：用户已排空作业。
- 正在更新…：正在更新作业。
- 已更新：用户已更新作业。
- 成功：作业已成功完成。
- 失败：作业未能完成。
如需查看一个流水线的详细信息，请点击该作业的名称。
访问作业监控图表
如需访问作业的监控图表，请在 Dataflow 监控界面中点击作业名称。这会将您导航到作业详情页面，其中包含以下内容：
- 作业图：流水线的直观展示
- 作业指标：关于作业执行情况的指标
- 作业信息面板：关于流水线的描述性信息
- 作业日志：Dataflow 服务在作业级层生成的日志
- 工作器日志：Dataflow 服务在工作器级层生成的日志
- 作业错误报告：显示沿所选时间轴发生的错误以及所有已记录错误计数的图表
- 时间选择器：可用于调整指标的时间范围的工具
在作业详情页面内，您可以使用作业图和作业指标标签页来切换作业视图。
要还原为旧版作业详情页面，请点击该页顶部的返回旧版作业页面 (Back to old job page)。
全屏模式
要以全屏模式查看指标图表，请点击 fullscreen。
使用时间选择器工具
您可以使用时间选择器工具来调整指标的时间范围。您可以选择预定义的持续时间或自定义时间间隔来分析作业。
对于流处理作业或运行中的批处理作业，默认显示的图表会显示相应作业前六个小时的指标。对于已停止或已完成的流处理作业，默认显示的图表会显示相应作业持续时间内的完整运行时间。
步骤和工作器指标
您可以查看如下指标的图表：
- 数据新鲜度（仅限流处理流水线）
- 系统延迟时间（仅限流处理流水线）
- 自动扩缩
- 吞吐量
- CPU 利用率
- 工作器错误日志计数
如需访问这些图表，请点击 Legend Toggle 以切换至“展开图表图例”。
吞吐量
吞吐量是指在任意时间点处理的数据量。该指标针对每个步骤计量，显示为每秒处理的元素数量。如需以“每秒字节数”为单位查看此指标，请点击吞吐量（元素数/秒） arrow_drop_down > 吞吐量（字节数/秒）。
CPU 利用率
CPU 利用率是用 CPU 的使用量除以可用于处理的 CPU 数量得到的值。该指标针对每个工作器计量，显示为百分比。
创建 Cloud Monitoring 提醒
Dataflow 与 Cloud Monitoring 完全集成，可让您在作业超出用户定义的阈值时创建提醒。
如需查看创建这些提醒的说明，请参阅使用 Cloud Monitoring for Dataflow 流水线页面。
查看流水线
当您选择特定 Dataflow 作业时，监控界面会显示该作业中流水线的详细信息。此信息包括流水线在 Dataflow 服务上执行时的图形表示、作业摘要、作业日志和流水线中每个步骤的详细信息。
Dataflow 监控界面可提供流水线的图形表示：即执行图。在流水线的执行图中，每个方框表示流水线中的一个转换。每个方框都包含转换名称和一些有关作业状态的信息，其中包括以下内容：
- 正在运行：该步骤目前正在运行。
- 已加入队列：FlexRS 作业中的步骤已加入队列。
- 成功：该步骤已成功完成。
- 已停止：由于作业停止，该步骤已停止。
- 未知：该步骤未能报告状态。
- 失败：该步骤未能完成。
基本执行图
流水线代码：Java：SDK 2.x
// Read the lines of the input text.
p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
// Count the words.
.apply(new CountWords())
// Write the formatted word counts to output.
.apply("WriteCounts", TextIO.write().to(options.getOutput()))
.to(options.getOutput()));
Python
(
p
# Read the lines of the input text.
| 'ReadLines' >> beam.io.ReadFromText(options.input)
# Count the words.
| CountWords()
# Write the formatted word counts to output.
| 'WriteCounts' >> beam.io.WriteToText(options.output))
Java：SDK 1.x |
执行图：
|
复合转换
在执行图中，您可以展开复合转换（即包含多个嵌套子转换的转换）。 展开式复合转换在执行图中以箭头标示。点击此箭头即可展开相关转换并查看其中所含的子转换。
流水线代码：Java：SDK 2.x
// The CountWords Composite Transform
// inside the WordCount pipeline.
public static class CountWords
extends PTransform<PCollection<String>, PCollection<String>> {
@Override
public PCollection<String> apply(PCollection<String> lines) {
// Convert lines of text into individual words.
PCollection<String> words = lines.apply(
ParDo.of(new ExtractWordsFn()));
// Count the number of times each word occurs.
PCollection<KV<String, Long>> wordCounts =
words.apply(Count.<String>perElement());
return wordCounts;
}
}
注意：右侧图片中的 Python
# The CountWords Composite Transform inside the WordCount pipeline.
class CountWords(beam.PTransform):
def expand(self, pcoll):
return (
pcoll
# Convert lines of text into individual words.
| 'ExtractWords' >> beam.ParDo(ExtractWordsFn())
# Count the number of times each word occurs.
| beam.combiners.Count.PerElement()
# Format each word and count into a printable string.
| 'FormatCounts' >> beam.ParDo(FormatCountsFn()))
Java：SDK 1.x |
执行图：
|
转换名称
Dataflow 可通过几种不同的方式来获取监控执行图中显示的转换名称：
Java：SDK 2.x
- Dataflow 可以使用您在应用转换时分配的名称。您为
apply方法提供的第一个参数将作为转换名称。 - Dataflow 可以推断转换名称：根据类名称（如果您已构建自定义转换）或
DoFn函数对象的名称（如果您使用的是ParDo等核心转换）进行推断。
Python
- Dataflow 可以使用您在应用转换时分配的名称。您可以通过指定转换的
label参数设置转换名称。 - Dataflow 可以推断转换名称：根据类名称（如果您已构建自定义转换）或
DoFn函数对象的名称（如果您使用的是ParDo等核心转换）进行推断。
Java：SDK 1.x
了解指标
实际用时
点击某个步骤时，系统会显示实际用时指标。实际用时提供了在所有工作器的所有线程中完成以下操作所花费的大致总时长：
- 初始化步骤
- 处理数据
- 重排数据
- 结束步骤
对于复合步骤，实际用时会告知您在各个分步骤中所花费时间的总和。此估算值可帮助您确定执行缓慢的步骤，以及诊断流水线中的哪个部分超出了预期执行时间。
请参阅了解 Dataflow 流水线中的耗时中的示例，了解如何使用实际用时（以前称为“总执行时间”）指标来调查导致流水线运行时间超出预期的原因。
侧边输入指标
侧边输入指标反映了侧边输入访问模式和算法对流水线性能的影响。如果您的流水线使用侧边输入，Dataflow 会将集合写入永久性层（如磁盘），而您的转换操作会从该永久性集合中读取内容。这些读取和写入操作会影响作业的执行时间。
如果您选择的转换会创建或使用侧边输入集合，Dataflow 监控界面便会显示侧边输入指标。您可以在步骤标签页的侧边输入指标部分中查看相关指标。
创建侧边输入的转换
如果所选转换创建了侧边输入集合，侧边输入指标部分便会显示集合名称以及如下指标：
- 写入操作用时：写入侧边输入集合所花的执行时间。
- 写入的字节数：已写入侧边输入集合的总字节数。
- 从侧边输入读取用时和字节数：一个包含更多指标的表，这些指标用于使用侧边输入集合的所有转换（称为“侧边输入使用者”）。
从侧边输入读取用时和字节数表包含各侧边输入使用者的以下信息：
- 侧边输入使用者：侧边输入使用者的转换名称。
- 读取操作用时：此使用者在读取侧边输入集合上所花的时间。
- 读取的字节数：此使用者从侧边输入集合中读取的字节数。
如果您的流水线包含一个可创建侧边输入的复合转换，请展开此复合转换，直到您看到创建该侧边输入的特定子转换。然后，选择该子转换以查看侧边输入指标部分。
图 5 显示了创建侧边输入集合的转换的侧边输入指标。
MakeMapView)。用于创建侧边输入的子转换 (CreateDataflowView) 处于选中状态，并且侧边输入指标显示在步骤标签页中。使用一个或多个侧边输入的转换
如果选择的转换使用一个或多个侧边输入，侧边输入指标部分便会显示从侧边输入读取用时和字节数表。此表会显示每个侧边输入集合的以下信息：
- 侧边输入集合：侧边输入集合的名称。
- 读取操作用时：转换在读取此侧边输入集合上所花的时间。
- 读取的字节数：转换从此侧边输入集合中读取的字节数。
如果您的流水线包含一个可读取侧边输入的复合转换，请展开此复合转换，直到您看到读取该侧边输入的特定子转换。然后，选择该子转换以查看侧边输入指标部分。
图 6 显示读取侧边输入集合的转换的侧边输入指标。
JoinBothCollections 转换用于读取侧边输入集合。在此执行图中，JoinBothCollections 处于选中状态，并且侧边输入指标显示在步骤标签页中。识别侧边输入性能问题
重复是常见的侧边输入性能问题。如果您的侧边输入 PCollection 过大，工作器将无法在内存中缓存整个集合。因此，工作器必须反复读取永久性侧边输入集合。
在图 7 中，侧边输入指标表明，从侧边输入集合读取的总字节数远远大于集合的实际大小（写入的总字节数）。
如需提升此流水线的性能，请重新设计算法，以避免循环访问或重新获取侧边输入数据。在此示例中，流水线生成两个集合的笛卡尔积。算法遍历整个侧边输入集合以获取主集合的各元素。您可以通过将主集合的多个元素一起进行批处理来改善管道的访问模式。这项更改可减少工作器必须重新读取侧边输入集合的次数。
如果您的流水线通过应用带有一个或多个大型侧边输入的 ParDo 来执行联接操作，则可能会出现另一个常见的性能问题。在此情况下，工作器会将连接操作的大部分执行时间用来读取侧边输入集合。
图 8 显示了此问题的侧边输入指标示例：
JoinBothCollections 转换的总执行时间为 18 分 31 秒。工作器将大部分执行时间（10 分 3 秒）用于读取 10 GB 的侧边输入集合。要提升此流水线的性能，请使用 CoGroupByKey 代替侧边输入。
系统延迟时间和数据新鲜度
在作业摘要面板中，您可以查看提供过去 6 小时的系统延迟时间和数据新鲜度指标的图表。系统每隔 60 秒对数据进行一次采样。采样后，数据在长达 180 秒的时间内不会显示。
系统延迟时间
系统延迟时间是指当前某数据项已处理或等待处理的最长时长。此指标表示元素在流水线中任意一个来源内等待的时间（以秒为单位）。处理完毕后，系统会调整最长时长。以下是其他注意事项：
- 如果有多个来源和接收器，则系统延迟时间是指元素在写入到所有接收器之前在来源内等待的最长时长。
- 如果某个来源没有提供元素在来源内等待多长时间的值，并且该元素没有用于定义其事件时间的元数据，则系统延迟时间会从流水线首次接收该元素的时间算起。
数据新鲜度
数据新鲜度是指实时与输出水印之间的时间长度。 流水线的每个步骤都有一个输出数据水印。输出数据水印 T 表示计算操作已处理事件时间早于 T 的所有元素。输出数据水印的上限为所有上游计算操作的最早输入数据水印。如果尚未处理某些输入数据，则输出水印可能会受到影响，进而影响数据新鲜度。
点击图表下方的创建提醒链接可打开 Cloud Monitoring 信息中心的新建提醒政策页面。
自动扩缩
在作业摘要面板中，您可以查看有关自动扩缩作业的信息。Dataflow 服务会自动选择运行您的自动扩缩作业所需的工作器实例数量。一段时间后，工作器实例数可能会随着作业的需求而发生变化。
要查看自动扩缩更改历史记录，请点击查看更多历史记录链接。此时会弹出一个窗口，其中包含流水线的工作器历史记录信息。
注意：您可以查看有关在 2016 年 12 月 12 日当天或之后运行或更新的流处理流水线的自动扩缩详细信息。如果您的流水线的运行时间或上次更新时间早于 12 月 12 日，那么只有在更新您的流水线后，您才能查看自动扩缩详细信息。
错误报告
Error Reporting 界面可汇总并显示流水线中产生的错误。
此错误报告包含以下内容：
- 带有错误消息的错误列表。
- 每个错误发生的次数。
- 直方图，指示每个错误发生的时间。
- 最近发生该错误的时间。
要查看流水线的错误报告，请依次点击流水线图表上方的日志菜单和流水线图表下方的堆栈追踪标签页。在 Dataflow 监控界面中，您将看到已记录的每个错误的摘要及其发生的次数。
工作器错误日志计数面板显示任何时间点在所有工作器中观察到的错误率。
如需详细了解错误，请点击错误摘要。您将转到 Error Reporting 界面。
Error Reporting 界面还提供了其他功能。请参阅查看错误，详细了解流水线产生的错误。