When you execute your pipeline using the Cloud Dataflow managed service, you can view that job and any others by using Cloud Dataflow's web-based monitoring user interface. The monitoring interface lets you see and interact with your Cloud Dataflow jobs.
You can access the Cloud Dataflow monitoring interface by using the Google Cloud Platform Console. The monitoring interface can show you:
- A list of all currently running Cloud 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 Platform 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 Cloud Dataflow monitoring interface:
- Log in to the GCP Console.
- Select your GCP project.
- Click the menu in the upper left corner.
- Navigate to the Big Data section and click Dataflow.
A list of Cloud 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 Cloud Dataflow quickstarts.
A job can have the following statuses:
- —: the monitoring UI has not yet received a status from the Cloud 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 Cloud 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 Cloud Dataflow service, as well as a job summary, the job log, and detailed information on each step in the pipeline.
The Cloud 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 complete.
- 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:
|
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: 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:
|
Transform names
Cloud Dataflow has a few different ways to obtain the transform name that's shown in the monitoring execution graph:
Java: SDK 2.x
- Cloud Dataflow can use a name that you assign when you apply your transform. The first
argument you supply to the
applymethod will be your transform name. - Cloud Dataflow can infer the transform name, either from the class name (if you've built a
custom transform) or the name of your
DoFnfunction object (if you're using a core transform such asParDo).
Python
- Cloud Dataflow can use a name that you assign when you apply your transform. You can set the
transform name by specifying the transform's
labelargument. - Cloud Dataflow can infer the transform name, either from the class name (if you've built a
custom transform) or the name of your
DoFnfunction object (if you're using a core transform such asParDo).
Java: SDK 1.x
- Cloud Dataflow can use a name that you assign when you apply your transform. The first
argument you supply to the
applymethod will be your transform name. - Cloud Dataflow can infer the transform name, either from the class name (if you've built a
custom transform) or the name of your
DoFnfunction 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.
See Understanding timing in Cloud 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, Cloud 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.
The Cloud 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.
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.
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.
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:
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.
Autoscaling
In the Job summary panel, you can view information about your autoscaling job. The Cloud 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 Cloud 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.
Hai bisogno di assistenza? Visita la nostra