Using the Cloud Dataflow monitoring interface

When you execute your pipeline using the Dataflow managed service, you can view that job and any others by using Dataflow's web-based monitoring user interface. The monitoring interface lets you see and interact with your Dataflow jobs.

Note: If you prefer to view and interact with your Dataflow jobs using the command-line, use the Dataflow command-line interface.

You can access the Dataflow monitoring interface by using the Google Cloud Console. The monitoring interface can show you:

• A list of all currently running Dataflow jobs and previously run jobs within the last 30 days.
• A graphical representation of each pipeline.
• Details about your job's status, execution, and SDK version.
• Links to information about the Google Cloud services running your pipeline, such as Compute Engine and Cloud Storage.
• Any errors or warnings that occur during a job.

Accessing the Cloud Dataflow monitoring interface

To access the Dataflow monitoring interface:

1. Log in to the Cloud Console.
2. Select your Google Cloud project.
3. Click the menu in the upper left corner.
4. Navigate to the Big Data section and click Dataflow.

A list of Dataflow jobs appears along with their status. If you don't see any jobs, you need to run a new job. To learn how to run a job, see the Dataflow quickstarts.

Figure 1: A list of Dataflow jobs in the Cloud Console with jobs in the Running, Failed, and Succeeded states.

A job can have the following statuses:

• —: the monitoring UI has not yet received a status from the Dataflow service.
• Running: the job is currently running.
• Not started: the job is created, but the system needs some time to prepare before launching.
• Queued: a FlexRS job is queued.
• Cancelling...: the job is being cancelled.
• Canceled: the job that was canceled by the user.
• Draining...: the job is being drained.
• Drained: the user drained the job.
• Updating...: the job is being updated.
• Updated: the user updated the job.
• Succeeded: the job has finished successfully.
• Failed: the job failed to complete.

For more information about a pipeline, select that job.

Viewing a pipeline

When you select a specific Dataflow job, the monitoring interface shows detailed information about the pipeline in that job. This information includes a graphical representation of your pipeline as it is executed on the Dataflow service, as well as a job summary, the job log, and detailed information on each step in the pipeline.

Figure 2: An individual pipeline job shown in the Dataflow monitoring interface.

The Dataflow monitoring interface provides a graphical representation of your pipeline: the execution graph. A pipeline's execution graph represents each transform in the pipeline as a box. Each box contains the transform name and information about the job status, which includes the following:

• Running: the step is currently running.
• Queued: the step in a FlexRS job is queued.
• Succeeded: the step finished successfully.
• Stopped: the step was stopped because the job stopped.
• Unknown: the step failed to report status.
• Failed: the step failed to complete.

Basic execution graph

Pipeline Code: 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 // Read the lines of the input text. p.apply(TextIO.Read.named("ReadLines").from(options.getInput())) // Count the words. .apply(new CountWords()) // Write the formatted word counts to output. .apply(TextIO.Write.named("WriteCounts") .to(options.getOutput()) .withNumShards(options.getNumShards()));

Execution Graph:

Figure 3: The pipeline code for a WordCount pipeline shown side-by-side with the resulting execution graph in the Dataflow monitoring interface.

Composite transforms

In the execution graph, composite transforms, transforms that contain multiple nested sub-transforms, are expandable. Expandable composite transforms are marked with an arrow in the graph. Click the arrow to expand the transform and view the sub-transforms within.

Pipeline Code: 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; } } Note: FormatCounts in the image to the right is not relevant to this SDK. 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 // 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()); // Format each word and count into a printable string. PCollection<String> results = wordCounts.apply( ParDo.of(new FormatCountsFn())); return results; } }

Execution Graph:

Figure 4: The pipeline code for the sub-steps of the CountWords transform shown side-by-side with the expanded execution graph for the entire pipeline.

Transform names

Dataflow has a few different ways to obtain the transform name that's shown in the monitoring execution graph:

Java: SDK 2.x

• Dataflow can use a name that you assign when you apply your transform. The first argument you supply to the apply method will be your transform name.
• Dataflow can infer the transform name, either from the class name (if you've built a custom transform) or the name of your DoFn function object (if you're using a core transform such as ParDo).

Python

• Dataflow can use a name that you assign when you apply your transform. You can set the transform name by specifying the transform's label argument.
• Dataflow can infer the transform name, either from the class name (if you've built a custom transform) or the name of your DoFn function object (if you're using a core transform such as ParDo).

Java: SDK 1.x

• Dataflow can use a name that you assign when you apply your transform. The first argument you supply to the apply method will be your transform name.
• Dataflow can infer the transform name, either from the class name (if you've built a custom transform) or the name of your DoFn function object (if you're using a core transform such as ParDo).

Wall time

When you click on a step, the Wall time metric shows up. Wall time provides the total approximate time spent across all threads in all workers on the following actions:

• Initializing the step
• Processing data
• Shuffling data
• Ending the step

For composite steps, wall time tells you the sum of time spent in the component steps. This estimate can help you identify slow steps and diagnose which part of your pipeline is taking more time than it should.

Figure 5: The Wall time metric can help you ensure your pipeline is running efficiently.

See Understanding timing in Dataflow pipelines for an example that uses the Wall Time metric (previously referred to as Total Execution Time) to investigate what is causing a pipeline to take more time to run than expected.

Side input metrics

Side input metrics show you how your side input access patterns and algorithms affect your pipeline's performance. When your pipeline uses a side input, Dataflow writes the collection to a persistent layer, such as a disk, and your transforms read from this persistent collection. These reads and writes affect your job's execution time.

Note: To see side input metrics, you must use Apache Beam SDK for Java version 2.3.0 or later. Apache Beam SDK for Python does not currently support side input metrics.

The Dataflow monitoring interface displays side input metrics when you select a transform that creates or consumes a side input collection. You can view the metrics in the Side Input Metrics section of the Step tab.

Transforms that create a side input

If the selected transform creates a side input collection, the Side Input Metrics section displays the name of the collection, along with the following metrics:

• Time spent writing: The execution time spent writing the side input collection.
• Bytes written: The total number of bytes written to the side input collection.
• Time & bytes read from side input: A table that contains additional metrics for all transforms that consume the side input collection, called side input consumers.

The Time & bytes read from side input table contains the following information for each side input consumer:

• Side input consumer: The transform name of the side input consumer.
• Time spent reading: The time this consumer spent reading the side input collection.
• Bytes read: The number of bytes this consumer read from the side input collection.

If your pipeline has a composite transform that creates a side input, expand the composite transform until you see the specific subtransform that creates the side input. Then, select that subtransform to view the Side Input Metrics section.

Figure 6 shows side input metrics for a transform that creates a side input collection.

Figure 6: The execution graph has an expanded composite transform (MakeMapView). The subtransform that creates the side input (CreateDataflowView) is selected, and the side input metrics are visible in the Step tab.

Transforms that consume one or more side inputs

If the selected transform consumes one or more side inputs, the Side Input Metrics section displays the Time & bytes read from side input table. This table contains the following information for each side input collection:

• Side input collection: The name of the side input collection.
• Time spent reading: The time the transform spent reading this side input collection.
• Bytes read: The number of bytes the transform read from this side input collection.

If your pipeline has a composite transform that reads a side input, expand the composite transform until you see the specific subtransform that reads the side input. Then, select that subtransform to view the Side Input Metrics section.

Figure 7 shows side input metrics for a transform that reads from an side input collection.

Figure 7: The JoinBothCollections transform reads from a side input collection. JoinBothCollections is selected in the execution graph and the side input metrics are visible in the Step tab.

Identifying side input performance issues

Reiteration is a common side input performance issue. If your side input PCollection is too large, workers can't cache the entire collection in memory. As a result, the workers must repeatedly read from the persistent side input collection.

In figure 8, side input metrics show that the total bytes read from the side input collection are much larger than the collection's size, total bytes written.

Figure 8: An example of reiteration. The side input collection is 563 MB, and the sum of the bytes read by consuming transforms is almost 12 GB.

To improve the performance of this pipeline, redesign your algorithm to avoid iterating or refetching the side input data. In this example, the pipeline creates the Cartesian product of two collections. The algorithm iterates through the entire side input collection for each element of the main collection. You can improve the access pattern of the pipeline by batching multiple elements of the main collection together. This change reduces the number of times workers must re-read the side input collection.

Another common performance issue can occur if your pipeline performs a join by applying a ParDo with one or more large side inputs. In this case, workers spend a large percentage of the join's execution time reading from the side input collections.

Figure 9 shows example side input metrics for this issue:

Figure 9: The JoinBothCollections transform has a total execution time of 18 min 31 sec. Workers spend the majority of the execution time (10 min 3 sec) reading from the 10 GB side input collection.

To improve the performance of this pipeline, use CoGroupByKey instead of side inputs.

System latency and data freshness

In the Job summary panel, you can view graphs that provide system latency and data freshness metrics for the last 6 hours. The data is sampled every 60 seconds. After sampling, data is not visible for up to 180 seconds.

System latency is the current maximum duration that an item of data has been awaiting processing. This metric indicates how long an element waits inside any one source in the pipeline, in seconds. Maximum duration is adjusted after processing. The following cases are additional considerations:

• If there are multiple sources and sinks, then system latency is the longest amount of time an element waits inside a source before it is written to all sinks.
• If a source does not provide a value for how long an element waits inside a source and the element does not have metadata to define its event time, then the system latency is calculated from the time when the pipeline first receives the element.

Data freshness is the amount of time since the event timestamp of the most recently fully-processed item of data. If there are multiple branches and sinks in the pipeline with data flowing through different paths, the data freshness is the maximum event timestamp of all elements.

Each step of your pipeline has an output data watermark. An output data watermark of T indicates that the computation has processed all elements with an event time before T. The output data watermark is bounded above by the earliest input data watermark of all upstream computations. If some input data has not yet been processed, the output watermark might be held back, which affects data freshness.

Clicking the Create alerting link below the graphs opens the Create new alerting policy page of the Stackdriver Monitoring dashboard.

Figure 10: View graphs with system latency and data freshness metrics.

Autoscaling

In the Job summary panel, you can view information about your autoscaling job. The Dataflow service automatically chooses the number of worker instances required to run your autoscaling job. The number of worker instances can change over time according to your job's needs.

To see the history of autoscaling changes, click the See More History link. A window with information about your pipeline's worker history will pop up.

Note: You can view autoscaling details about streaming pipelines run or updated on or after December 12, 2016. If your pipeline was run or last updated prior to December 12, you will only be able to see autoscaling details after you update your pipeline.

Error Reporting

The Stackdriver error reporting interface aggregates and displays errors produced in your pipelines.

The error report includes:

• A list of errors with error messages.
• The number of times each error occurred.
• A histogram indicating when each error occurred.
• The time that the error most recently occurred.

To view the error report for your pipeline, click the Logs menu above the pipeline graph and then on the Stack Traces tab below the pipeline graph. In the Dataflow monitoring interface, you will see a summary of each logged error and the number of times it occurred.

To see more information about the errors, click an error summary. You will be taken to the Stackdriver error reporting interface.

The Stackdriver error reporting interface offers additional functionality. See Viewing Errors for more information about the errors produced by your pipelines.