Deploying a pipeline

This document explains in detail how Dataflow deploys and runs a pipeline, and covers advanced topics like optimization and load balancing. If you are looking for a step-by-step guide on how to create and deploy your first pipeline, use Dataflow's quickstarts for Java, Python or templates.

After you construct and test your Apache Beam pipeline, you can use the Dataflow managed service to deploy and execute it. Once on the Dataflow service, your pipeline code becomes a Dataflow job.

The Dataflow service fully manages Google Cloud services such as Compute Engine and Cloud Storage to run your Dataflow job, automatically spinning up and tearing down the necessary resources. The Dataflow service provides visibility into your job through tools like the Dataflow monitoring interface and the Dataflow command-line interface.

You can control some aspects of how the Dataflow service runs your job by setting execution parameters in your pipeline code. For example, the execution parameters specify whether the steps of your pipeline run on worker virtual machines, on the Dataflow service backend, or locally.

In addition to managing Google Cloud resources, the Dataflow service automatically performs and optimizes many aspects of distributed parallel processing. These include the following:

• Parallelization and distribution. Dataflow automatically partitions your data and distributes your worker code to Compute Engine instances for parallel processing. For more information, see parallelization and distribution.
• Optimization. Dataflow uses your pipeline code to create an execution graph that represents your pipeline's PCollections and transforms, and optimizes the graph for the most efficient performance and resource usage. Dataflow also automatically optimizes potentially costly operations, such as data aggregations. For more information, see Fusion optimization and Combine optimization.
• Automatic tuning features. The Dataflow service includes several features that provide on-the-fly adjustment of resource allocation and data partitioning, such as Horizontal Autoscaling, Vertical Autoscaling, and Dynamic Work Rebalancing. These features help the Dataflow service execute your job as quickly and efficiently as possible.

Pipeline lifecycle: from pipeline code to Dataflow job

When you run your Dataflow pipeline, Dataflow creates an execution graph from the code that constructs your Pipeline object, including all of the transforms and their associated processing functions (such as DoFns). This phase is called Graph Construction Time and runs locally on the computer where the pipeline is run.

During graph construction, Apache Beam locally executes the code from the main entry point of the pipeline code, stopping at the calls to a source, sink or transform step, and turning these calls into nodes of the graph. As a consequence, a piece of code in a pipeline's entry point (Java's main() method or the top-level of a Python script) locally executes on the machine that runs the pipeline, while the same code declared in a method of a DoFn object executes in the Dataflow workers.

Also during graph construction, Apache Beam validates that any resources referenced by the pipeline (like Cloud Storage buckets, BigQuery tables, and Pub/Sub topics or subscriptions) actually exist and are accessible. The validation is done through standard API calls to the respective services, so it's vital that the user account used to run a pipeline has proper connectivity to the necessary services and is authorized to call their APIs. Before submitting the pipeline to the Dataflow service, Apache Beam also checks for other errors, and ensures that the pipeline graph doesn't contain any illegal operations.

The execution graph is then translated into JSON format, and the JSON execution graph is transmitted to the Dataflow service endpoint.

Note: Graph construction also happens when you execute your pipeline locally, but the graph is not translated to JSON or transmitted to the service. Instead, the graph is run locally on the same machine where you launched your Dataflow program. See the documentation on configuring for local execution for more details.

The Dataflow service then validates the JSON execution graph. When the graph is validated, it becomes a job on the Dataflow service. You'll be able to see your job, its execution graph, status, and log information by using the Dataflow monitoring interface.

Java

The Dataflow service sends a response to the machine where you ran your Dataflow program. This response is encapsulated in the object DataflowPipelineJob, which contains your Dataflow job's jobId. You can use the jobId to monitor, track, and troubleshoot your job using the Dataflow monitoring interface and the Dataflow command-line interface. See the API reference for DataflowPipelineJob for more information.

Python

The Dataflow service sends a response to the machine where you ran your Dataflow program. This response is encapsulated in the object DataflowPipelineResult, which contains your Dataflow job's job_id. You can use the job_id to monitor, track, and troubleshoot your job using the Dataflow monitoring interface and the Dataflow command-line interface.

Execution graph

Dataflow builds a graph of steps that represents your pipeline, based on the transforms and data you used when you constructed your Pipeline object. This is the pipeline execution graph.

The WordCount example, included with the Apache Beam SDKs, contains a series of transforms to read, extract, count, format, and write the individual words in a collection of text, along with an occurrence count for each word. The following diagram shows how the transforms in the WordCount pipeline are expanded into an execution graph:

Figure 1: WordCount Example Execution Graph

The execution graph often differs from the order in which you specified your transforms when you constructed the pipeline. This is because the Dataflow service performs various optimizations and fusions on the execution graph before it runs on managed cloud resources. The Dataflow service respects data dependencies when executing your pipeline; however, steps without data dependencies between them may be executed in any order.

You can see the unoptimized execution graph that Dataflow has generated for your pipeline when you select your job in the Dataflow monitoring interface. For more information about viewing jobs, see Using the Dataflow monitoring interface.

Parallelization and distribution

The Dataflow service automatically parallelizes and distributes the processing logic in your pipeline to the workers you've allotted to perform your job. Dataflow uses the abstractions in the programming model to represent parallel processing functions; for example, your ParDo transforms cause Dataflow to automatically distribute your processing code (represented by DoFns) to multiple workers to be run in parallel.

Structuring your user code

You can think of your DoFn code as small, independent entities: there can potentially be many instances running on different machines, each with no knowledge of the others. As such, pure functions (functions that do not depend on hidden or external state, that have no observable side effects, and are deterministic) are ideal code for the parallel and distributed nature of DoFns.

The pure function model is not strictly rigid, however; state information or external initialization data can be valid for DoFn and other function objects, so long as your code does not depend on things that the Dataflow service does not guarantee. When structuring your ParDo transforms and creating your DoFns, keep the following guidelines in mind:

• The Dataflow service guarantees that every element in your input PCollection is processed by a DoFn instance exactly once.
• The Dataflow service does not guarantee how many times a DoFn will be invoked.
• The Dataflow service does not guarantee exactly how the distributed elements are grouped—that is, it does not guarantee which (if any) elements are processed together.
• The Dataflow service does not guarantee the exact number of DoFn instances that will be created over the course of a pipeline.
• The Dataflow service is fault-tolerant, and may retry your code multiple times in the case of worker issues. The Dataflow service may create backup copies of your code, and can have issues with manual side effects (such as if your code relies upon or creates temporary files with non-unique names).
• The Dataflow service serializes element processing per DoFn instance. Your code does not need to be strictly thread-safe; however, any state shared between multiple DoFn instances must be thread-safe.

See Requirements for User-Provided Functions in the programming model documentation for more information about building your user code.

Error and exception handling

Your pipeline may throw exceptions while processing data. Some of these errors are transient (e.g., temporary difficulty accessing an external service), but some are permanent, such as errors caused by corrupt or unparseable input data, or null pointers during computation.

Dataflow processes elements in arbitrary bundles, and retries the complete bundle when an error is thrown for any element in that bundle. When running in batch mode, bundles including a failing item are retried 4 times. The pipeline will fail completely when a single bundle has failed 4 times. When running in streaming mode, a bundle including a failing item will be retried indefinitely, which may cause your pipeline to permanently stall.

Startup worker errors, like failure to install packages on the workers, are transient, which will result in indefinite retries and may cause your pipeline to permanently stall.

Note: When processing in batch mode, you might see a large number of individual failures before a pipeline job fails completely (which happens when any given bundle fails after four retry attempts). For example, if your pipeline attempts to process 100 bundles, Dataflow could theoretically generate several hundred individual failures until a single bundle reaches the 4-failure condition for exit.

Fusion optimization

Once the JSON form of your pipeline's execution graph has been validated, the Dataflow service may modify the graph to perform optimizations. Such optimizations can include fusing multiple steps or transforms in your pipeline's execution graph into single steps. Fusing steps prevents the Dataflow service from needing to materialize every intermediate PCollection in your pipeline, which can be costly in terms of memory and processing overhead.

While all the transforms you've specified in your pipeline construction are executed on the service, they may be executed in a different order, or as part of a larger fused transform to ensure the most efficient execution of your pipeline. The Dataflow service respects data dependencies between the steps in the execution graph, but otherwise steps may be executed in any order.

Fusion example

The following diagram shows how the execution graph from the WordCount example included with the Apache Beam SDK for Java might be optimized and fused by the Dataflow service for efficient execution:

Figure 2: WordCount Example Optimized Execution Graph

Preventing fusion

There are a few cases in your pipeline where you may want to prevent the Dataflow service from performing fusion optimizations. These are cases in which the Dataflow service might incorrectly guess the optimal way to fuse operations in the pipeline, which could limit the Dataflow service's ability to make use of all available workers.

For example, one case in which fusion can limit Dataflow's ability to optimize worker usage is a "high fan-out" ParDo. In such an operation, you might have an input collection with relatively few elements, but the ParDo produces an output with hundreds or thousands of times as many elements, followed by another ParDo. If the Dataflow service fuses these ParDo operations together, parallelism in this step is limited to at most the number of items in the input collection, even though the intermediate PCollection contains many more elements.

You can prevent such a fusion by adding an operation to your pipeline that forces the Dataflow service to materialize your intermediate PCollection. Consider using one of the following operations:

• You can insert a GroupByKey and ungroup after your first ParDo. The Dataflow service never fuses ParDo operations across an aggregation.
• You can pass your intermediate PCollection as a side input to another ParDo. The Dataflow service always materializes side inputs.
• You can insert a Reshuffle step. Reshuffle prevents fusion, checkpoints the data, and performs deduplication of records. Reshuffle is supported by Dataflow even though it is marked deprecated in the Apache Beam documentation.

Monitoring fusion

You can access your optimized graph and fused stages by calling project.locations.jobs.get or by running the following gcloud command:

  gcloud dataflow jobs describe --full $JOB_ID --format json

The fused stages are described under the ExecutionStageSummary object within the ComponentTransform array on the output response file. You can consider piping the output to jq to easily extract the relevant bits using the following gcloud command:

  gcloud dataflow jobs describe --full $JOB_ID --format json | jq '.pipelineDescription.executionPipelineStage[] | {"stage_id": .id, "stage_name": .name, "fused_steps": .componentTransform }'

Combine optimization

Aggregation operations are an important concept in large-scale data processing. Aggregation brings together data that's conceptually far apart, making it extremely useful for correlating. The Dataflow programming model represents aggregation operations as the GroupByKey, CoGroupByKey, and Combine transforms.

Dataflow's aggregation operations combine data across the entire data set, including data that may be spread across multiple workers. During such aggregation operations, it's often most efficient to combine as much data locally as possible before combining data across instances. When you apply a GroupByKey or other aggregating transform, the Dataflow service automatically performs partial combining locally before the main grouping operation.

Note: Because the Dataflow service automatically performs partial local combining, it is strongly recommended that you do not attempt to make this optimization by hand in your pipeline code.

When performing partial or multi-level combining, the Dataflow service makes different decisions based on whether your pipeline is working with batch or streaming data. For bounded data, the service favors efficiency and will perform as much local combining as possible. For unbounded data, the service favors lower latency, and may not perform partial combining (as it may increase latency).

Autotuning features

The Dataflow service contains several autotuning features that can further dynamically optimize your Dataflow job while it is running. These features include Horizontal Autoscaling, Vertical Autoscaling, and Dynamic Work Rebalancing.

Note: With the introduction of Vertical Autoscaling, the feature that was previously called Autoscaling is now called Horizontal Autoscaling.

Horizontal Autoscaling

With Horizontal Autoscaling enabled, the Dataflow service automatically chooses the appropriate number of worker instances required to run your job. The Dataflow service may also dynamically re-allocate more workers or fewer workers during runtime to account for the characteristics of your job. Certain parts of your pipeline may be computationally heavier than others, and the Dataflow service may automatically spin up additional workers during these phases of your job (and shut them down when they're no longer needed).

Note: In jobs running Dataflow Prime, Horizontal Autoscaling (previously called Autoscaling) is deactivated during and up to 10 minutes after Vertical Autoscaling. For more information, read about the effect of Vertical Autoscaling on Horizontal Autoscaling.

Java

Horizontal Autoscaling is enabled by default on the streaming jobs that use Streaming Engine and all batch jobs. You can disable Horizontal Autoscaling by specifying the flag --autoscalingAlgorithm=NONE when you run your pipeline; if so, note that the Dataflow service sets the number of workers based on the --numWorkers option, which defaults to 3.

With Horizontal Autoscaling enabled, the Dataflow service does not allow user control of the exact number of worker instances allocated to your job. You might still cap the number of workers by specifying the --maxNumWorkers option when you run your pipeline.

For batch jobs, the --maxNumWorkers flag is optional. The default is 1000. For streaming jobs using Streaming Engine, the --maxNumWorkers flag is optional. The default is 100. For streaming jobs not using Streaming Engine, the --maxNumWorkers flag is required.

Python

Horizontal Autoscaling is enabled by default on all batch Dataflow jobs created using the Apache Beam SDK for Python version 0.5.1 or later. You can disable Horizontal Autoscaling by explicitly specifying the flag --autoscaling_algorithm=NONE when you run your pipeline; if so, note that the Dataflow service sets the number of workers based on the --num_workers option, which defaults to 3.

With Horizontal Autoscaling enabled, the Dataflow service doesn't let you control the exact number of worker instances allocated to your job. You might still cap the number of workers by specifying the --max_num_workers option when you run your pipeline.

For batch jobs, the --max_num_workers flag is optional. The default is 1000. For streaming jobs using Streaming Engine, the --max_num_workers flag is optional. The default is 100. For streaming jobs not using Streaming Engine, the --max_num_workers flag is required.

Dataflow scales based on the parallelism of a pipeline. The parallelism of a pipeline is an estimate of the number of threads needed to most efficiently process data at any given time.

Batch autoscaling

For batch pipelines, Dataflow automatically chooses the number of workers based on the estimated total amount of work in each stage of your pipeline, which is dependent on both the input size and the current throughput. Dataflow re-evaluates the amount of work according to the progress of the execution every 30 seconds and dynamically scales up or down the number of workers as estimated total amount of work increases or decreases.

Note: The number of workers is sub-linear to the amount of work. For instance, a job with twice the work has less than twice the workers.

If any of the following conditions occur, to save idle resources, Dataflow either maintains or decreases the number of workers:

Average worker CPU usage is lower than 5%.

Parallelism is limited due to unparallelizable work, such as un-splittable data like compressed files or data processed by I/O modules that don't split.

Number of parallelism is fixed, such as writing to existing files at a Cloud Storage destination.

Note: For pipelines not using Dataflow Shuffle, Dataflow might not be able to scale down the workers effectively because the workers may have shuffled data stored in local disks.

If your pipeline uses a custom data source that you've implemented, there are a few methods you can implement that provide more information to the Dataflow service's Horizontal Autoscaling algorithm and potentially improve performance:

Java

• In your BoundedSource subclass, implement the method getEstimatedSizeBytes. The Dataflow service uses getEstimatedSizeBytes when calculating the initial number of workers to use for your pipeline.
• In your BoundedReader subclass, implement the method getFractionConsumed. The Dataflow service uses getFractionConsumed to track read progress and converge on the correct number of workers to use during a read.

Python

• In your BoundedSource subclass, implement the method estimate_size. The Dataflow service uses estimate_size when calculating the initial number of workers to use for your pipeline.
• In your RangeTracker subclass, implement the method fraction_consumed. The Dataflow service uses fraction_consumed to track read progress and converge on the correct number of workers to use during a read.

Streaming autoscaling

Note: Streaming autoscaling is generally available for pipelines that use Streaming Engine. For pipelines that do not use Streaming Engine, streaming autoscaling is available in beta.

For more information about launch stage definitions, see the Product launch stages page.

Streaming autoscaling allows the Dataflow service to adaptively change the number of workers used to execute your streaming pipeline in response to changes in load and resource utilization. Streaming autoscaling is offered at no charge and is designed to reduce the costs of the resources used when executing streaming pipelines.

The streaming autoscaling determines when to scale by monitoring the estimated backlog time. The estimated backlog time is calculated from both the throughput and the backlog bytes still to be processed from the input source. A pipeline is considered 'backlogged' when the estimated backlog time stays above 15 seconds.

Scaling up: If a streaming pipeline remains backlogged with workers utilizing, on average, more than 20% of their CPUs, for a couple minutes, Dataflow scales up. Dataflow targets clearing the backlog in approximately 150 seconds after scaling up, given the current throughput per worker.

Scaling down: If a streaming pipeline backlog is lower than 10 seconds and workers are utilizing on average less than 75% of the CPUs for a period of a couple minutes, Dataflow scales down. After scaling down, workers utilize on average, 75% of their CPUs. In streaming jobs that do not use Streaming Engine, sometimes the 75% CPU utilization cannot be achieved due to disk distribution (each worker must have the same number of persistent disks), and a lower CPU utilization is used. For example, a job set to use a maximum of 100 workers (with 1 disk per worker) can be scaled down to 50 workers (with 2 disks per worker). For this job, a 75% CPU utilization is not achievable because the next scale down from 100 workers is 50 workers, which is less than the required 75 workers. Consequently, Dataflow does not scale down this job, resulting in a lower than 75% CPU utilization.

No scaling: If there is no backlog but CPU usage is 75% or greater, the pipeline does not scale down. If there is backlog but CPU usage is less than 20% the pipeline does not scale up.

Predictive scaling: The streaming engine also uses a predictive Horizontal Autoscaling technique based on timer backlog. In the Dataflow model, the unbounded data in a streaming pipeline is divided into windows grouped by timestamps. At the end of a window, timers fire for each key being processed in that window. The firing of a timer indicates that the window is expired for a given key. The streaming engine is able to measure the timer backlog, meaning that it can predict how many timers are going to fire at the end of a window. Using the timer backlog as a signal lets Dataflow "see the future" by estimating the amount of processing that will need to happen when future timers fire. Based on the estimated future load, Dataflow autoscales in advance to meet expected demand.

Without Horizontal Autoscaling, you choose a fixed number of workers by specifying numWorkers or num_workers to execute your pipeline. As the input workload varies over time, this number can become either too high or too low. Provisioning too many workers results in unnecessary extra cost, and provisioning too few workers results in higher latency for processed data. By enabling Horizontal Autoscaling, resources are used only as they are needed.

The objective of Horizontal Autoscaling streaming pipelines is to minimize backlog while maximizing worker utilization and throughput, and quickly react to spikes in load. By enabling Horizontal Autoscaling, you don't have to choose between provisioning for peak load and fresh results. Workers are added as CPU utilization and backlog increase and are removed as these metrics come down. This way, you’re paying only for what you need, and the job is processed as efficiently as possible.

Java

Custom unbounded sources

If your pipeline uses a custom unbounded source, the source must inform the Dataflow service about backlog. Backlog is an estimate of the input in bytes that has not yet been processed by the source. To inform the service about backlog, implement either one of the following methods in your UnboundedReader class.

• getSplitBacklogBytes() - Backlog for the current split of the source. The service aggregates backlog across all the splits.
• getTotalBacklogBytes() - The global backlog across all the splits. In some cases the backlog is not available for each split and can only be calculated across all the splits. Only the first split (split ID '0') needs to provide total backlog.

The Apache Beam repository contains several examples of custom sources that implement the UnboundedReader class.

Enable streaming autoscaling

For streaming jobs using Streaming Engine, Horizontal Autoscaling is enabled by default.

To enable Horizontal Autoscaling for jobs not using Streaming Engine, set the following execution parameters when you start your pipeline:

--autoscalingAlgorithm=THROUGHPUT_BASED
--maxNumWorkers=N

For streaming jobs not using Streaming Engine, the minimum number of workers is 1/15th of the --maxNumWorkers value, rounded up.

Streaming pipelines are deployed with a fixed pool of Persistent Disks, equal in number to --maxNumWorkers. Take this into account when you specify --maxNumWorkers, and ensure this value is a sufficient number of disks for your pipeline.

Note: If you've reached a scaling limit and want to raise the --maxNumWorkers, you must submit a new job with a higher --maxNumWorkers.

If you want to update a streaming autoscaling job that's not using Streaming Engine, make sure --maxNumWorkers remains the same (see the section on manually scaling streaming pipelines). Not specifying the --autoscalingAlgorithm pipeline option in the Update command disables Horizontal Autoscaling for the updated job.

Usage and pricing

Compute Engine usage is based on the average number of workers, while Persistent Disk usage is based on the exact value of --maxNumWorkers. Persistent Disks are redistributed such that each worker gets an equal number of attached disks.

In the example above, where --maxNumWorkers=15, you pay for between 1 and 15 Compute Engine instances and exactly 15 Persistent Disks.

Python

Enable streaming autoscaling

To enable Horizontal Autoscaling, set the following execution parameters when you start your pipeline:

--autoscaling_algorithm=THROUGHPUT_BASED
--max_num_workers=N

For streaming jobs not using Streaming Engine, the minimum number of workers is 1/15th of the --maxNumWorkers value, rounded up.

Streaming pipelines are deployed with a fixed pool of Persistent Disks, equal in number to --maxNumWorkers. Take this into account when you specify --maxNumWorkers, and ensure this value is a sufficient number of disks for your pipeline.

Note: If you've reached a scaling limit and want to raise the --maxNumWorkers, you must submit a new job with a higher --maxNumWorkers.

If you want to update a streaming autoscaling job that's not using Streaming Engine, make sure --maxNumWorkers remains the same (see the section on manually scaling streaming pipelines). Not specifying the --autoscalingAlgorithm pipeline option in the Update command disables Horizontal Autoscaling for the updated job.

Usage and pricing

Compute Engine usage is based on the average number of workers, while Persistent Disk usage is based on the exact value of --max_num_workers. Persistent Disks are redistributed such that each worker gets an equal number of attached disks.

In the example above, where --max_num_workers=15, you pay for between 1 and 15 Compute Engine instances and exactly 15 Persistent Disks.

Manually scaling a streaming pipeline

Until Horizontal Autoscaling is generally available in streaming mode, there is a workaround you can use to manually scale the number of workers running your streaming pipeline by using Dataflow's Update feature.

Java

To scale your streaming pipeline during execution, ensure that you set the following execution parameters when you start your pipeline:

• Set --maxNumWorkers equal to the maximum number of workers you want available to your pipeline.
• Set --numWorkers equal to the initial number of workers you want your pipeline to use when it starts running.

Once your pipeline is running, you can Update your pipeline and specify a new number of workers using the --numWorkers parameter. The value you set for the new --numWorkers must be between N and --maxNumWorkers, where N is equal to --maxNumWorkers / 15.

Updating your pipeline replaces your running job with a new job, using the new number of workers, while preserving all state information associated with the previous job.

Note: Your pipeline's maximum scaling range depends on the number of persistent disks deployed when the pipeline starts. The Dataflow service deploys one persistent disk per worker at the maximum number of workers. Deploying extra Persistent Disks by setting --maxNumWorkers to a higher value than --numWorkers provides some benefits to your pipeline—specifically, it allows you the flexibility to scale your pipeline to a larger number of workers after startup, and might provide improved performance. However, your pipeline might also incur additional cost for the extra Persistent Disks. Take note of the cost and quota implications of the additional Persistent Disk resources when planning your streaming pipeline and setting the scaling range.

Note: You cannot change the scaling range of a pipeline by using the Update feature. If you need to scale further, you must start a new pipeline and specify a higher value for --maxNumWorkers as the ceiling of your desired scaling range.

Python

To scale your streaming pipeline during execution, ensure that you set the following execution parameters when you start your pipeline:

• Set --max_num_workers equal to the maximum number of workers you want available to your pipeline.
• Set --num_workers equal to the initial number of workers you want your pipeline to use when it starts running.

Once your pipeline is running, you can Update your pipeline and specify a new number of workers using the --num_workers parameter. The value you set for the new --num_workers must be between N and --max_num_workers, where N is equal to --max_num_workers / 15.

Updating your pipeline replaces your running job with a new job, using the new number of workers, while preserving all state information associated with the previous job.

Note: Your pipeline's maximum scaling range depends on the number of Persistent Disks deployed when the pipeline starts. The Dataflow service deploys one persistent disk per worker at the maximum number of workers. Deploying extra persistent disks by setting --max_num_workers to a higher value than --num_workers provides some benefits to your pipeline—specifically, it allows you the flexibility to scale your pipeline to a larger number of workers after startup, and might provide improved performance. However, your pipeline might also incur additional cost for the extra Persistent Disks. Take note of the cost and quota implications of the additional Persistent Disk resources when planning your streaming pipeline and setting the scaling range.

Note: You cannot change the scaling range of a pipeline by using the Update feature. If you need to scale further, you must start a new pipeline and specify a higher value for --max_num_workers as the ceiling of your desired scaling range.

Vertical Autoscaling for Python streaming pipelines

Preview — Dataflow Prime

This feature is covered by the Pre-GA Offerings Terms of the Google Cloud Terms of Service. Pre-GA features might have limited support, and changes to pre-GA features might not be compatible with other pre-GA versions. For more information, see the launch stage descriptions.

Vertical Autoscaling is a feature that enables Dataflow Prime to dynamically scale up or down the memory available to workers to fit the requirements of the job. The feature is designed to make jobs resilient to out-of-memory errors, and to maximize pipeline efficiency. Dataflow Prime monitors your pipeline, detects situations where the workers lack or exceed available memory, and then replaces them with new workers with more or less memory.

Vertical Autoscaling is enabled when you enable Dataflow Prime.

Important: Because Vertical Autoscaling replaces existing workers with new workers, we strongly recommend using custom containers to improve the latency that might arise from resizing the workers.

Preview limitations

• Only Python streaming jobs can be vertically scaled.
• Only the memory of the workers' VMs can be vertically scaled.
• Memory upscaling has an upper limit of 8 GiB for each vCPU. While using GPUs, the upper limit for memory upscaling is 6.5 GiB for each vCPU.
• Memory downscaling has a lower limit of 3 GiB for each vCPU.

Monitoring Vertical Autoscaling

Vertical Autoscaling operations are published to the job and worker logs. To view these logs, see Using the Dataflow monitoring interface.

Effect on Horizontal Autoscaling

In Dataflow Prime, Vertical Autoscaling works alongside Horizontal Autoscaling. This combination enables Dataflow Prime to seamlessly scale workers up or down to best fit the needs of your pipeline and maximize the utilization of the compute capacity.

By design, Vertical Autoscaling (which adjusts the VM memory) occurs at a lower frequency than Horizontal Autoscaling (which adjusts the number of VMs). Horizontal Autoscaling is deactivated during and up to 10 minutes after an update is triggered by Vertical Autoscaling. If there exists a significant backlog of input data after this 10-minute mark, Horizontal Autoscaling is likely to occur to clear that backlog. To learn about Horizontal Autoscaling for streaming pipelines, see Streaming autoscaling.

Troubleshooting

This section provides instructions for troubleshooting common issues related to vertical autoscaling.

Vertical Autoscaling does not seem to work. What should I check?

• Verify that your job is a Python streaming job. Vertical Autoscaling is not available for other job types.
• Verify that you enabled the Cloud Autoscaling API for your Google Cloud project.
• Verify that your job is running Dataflow Prime. For more information, see Enabling Dataflow Prime.

Job observes high backlog and high watermark. What should I check?

If the vertical reshaping of workers takes longer than a few minutes, then your job might exhibit a high backlog of the input data and a high watermark. To address this issue, we strongly recommend that you use custom containers as they can improve the latency that might arise from reshaping the workers. If this issue persists after using custom containers, contact Customer Care.

Dynamic Work Rebalancing

The Dataflow service's Dynamic Work Rebalancing feature allows the service to dynamically re-partition work based on runtime conditions. These conditions might include the following:

• Imbalances in work assignments
• Workers taking longer than expected to finish
• Workers finishing faster than expected

The Dataflow service automatically detects these conditions and can dynamically reassign work to unused or underused workers to decrease your job's overall processing time.

Limitations

Dynamic work rebalancing only happens when the Dataflow service is processing some input data in parallel: when reading data from an external input source, when working with a materialized intermediate PCollection, or when working with the result of an aggregation like GroupByKey. If a large number of steps in your job are fused, there are fewer intermediate PCollections in your job and dynamic work rebalancing will be limited to the number of elements in the source materialized PCollection. If you want to ensure that dynamic work rebalancing can be applied to a particular PCollection in your pipeline, you can prevent fusion in a few different ways to ensure dynamic parallelism.

Dynamic work rebalancing cannot re-parallelize data finer than a single record. If your data contains individual records that cause large delays in processing time, they may still delay your job, since Dataflow cannot subdivide and redistribute an individual "hot" record to multiple workers.

Java

If you've set a fixed number of shards for your pipeline's final output (for example, by writing data using TextIO.Write.withNumShards), parallelization will be limited based on the number of shards that you've chosen.

Python

If you've set a fixed number of shards for your pipeline's final output (for example, by writing data using beam.io.WriteToText(..., num_shards=...)), Dataflow will limit parallelization based on the number of shards that you've chosen.

The fixed-shards limitation can be considered temporary, and may be subject to change in future releases of the Dataflow service.

Working with Custom Data Sources

Java

If your pipeline uses a custom data source that you provide, you must implement the method splitAtFraction to allow your source to work with the dynamic work rebalancing feature.

Caution: Using dynamic work rebalancing with custom data sources is an extremely advanced use case. If you choose to implement splitAtFraction, it is critical that you test your code extensively and with maximum code coverage.

If you implement splitAtFraction incorrectly, records from your source may appear to get duplicated or dropped. See the API reference information on RangeTracker for help and tips on implementing splitAtFraction.

Python

If your pipeline uses a custom data source that you provide, your RangeTracker must implement try_claim, try_split, position_at_fraction, and fraction_consumed to allow your source to work with the dynamic work rebalancing feature.

See the API reference information on RangeTracker for more information.

Resource usage and management

The Dataflow service fully manages resources in Google Cloud on a per-job basis. This includes spinning up and shutting down Compute Engine instances (occasionally referred to as workers or VMs) and accessing your project's Cloud Storage buckets for both I/O and temporary file staging. However, if your pipeline interacts with Google Cloud data storage technologies like BigQuery and Pub/Sub, you must manage the resources and quota for those services.

Dataflow uses a user provided location in Cloud Storage specifically for staging files. This location is under your control, and you should ensure that the location's lifetime is maintained as long as any job is reading from it. You can re-use the same staging location for multiple job runs, as the SDK's built-in caching can speed up the start time for your jobs.

Caution: Manually altering Dataflow-managed Compute Engine resources associated with a Dataflow job is an unsupported operation. You should not attempt to manually stop, delete, or otherwise control the Compute Engine instances that Dataflow has created to run your job. In addition, you should not alter any persistent disk resources associated with your Dataflow job.

Jobs

You may run up to 25 concurrent Dataflow jobs per Google Cloud project; however, this limit can be increased by contacting Google Cloud Support. For more information, see Quotas.

The Dataflow service is currently limited to processing JSON job requests that are 20 MB in size or smaller. The size of the job request is specifically tied to the JSON representation of your pipeline; a larger pipeline means a larger request.

To estimate the size of your pipeline's JSON request, run your pipeline with the following option:

Java

--dataflowJobFile=< path to output file >

Python

--dataflow_job_file=< path to output file >

This command writes a JSON representation of your job to a file. The size of the serialized file is a good estimate of the size of the request; the actual size will be slightly larger due to some additional information included in the request.

For more information, see the troubleshooting page for "413 Request Entity Too Large" / "The size of serialized JSON representation of the pipeline exceeds the allowable limit".

In addition, your job's graph size must not exceed 10 MB. For more information, see the troubleshooting page for "The job graph is too large. Please try again with a smaller job graph, or split your job into two or more smaller jobs.".

Workers

The Dataflow service currently allows a maximum of 1000 Compute Engine instances per job. For batch jobs, the default machine type is n1-standard-1. For streaming jobs, the default machine type for Streaming Engine-enabled jobs is n1-standard-2 and the default machine type for non-Streaming Engine jobs is n1-standard-4. When using the default machine types, the Dataflow service can therefore allocate up to 4000 cores per job. If you need more cores for your job, you can select a larger machine type.

Note: The Dataflow managed service now deploys Compute Engine virtual machines associated with Dataflow jobs using Managed Instance Groups. A Managed Instance Group creates multiple Compute Engine instances from a common template and allows you to control and manage them as a group. That way, you don't have to individually control each instance associated with your pipeline.

You should not attempt to manage or otherwise interact directly with your Compute Engine Managed Instance Group; the Dataflow service will take care of that for you. Manually altering any Compute Engine resources associated with your Dataflow job is an unsupported operation.

You can use any of the available Compute Engine machine type families as well as custom machine types. For best results, use n1 machine types. Shared core machine types, such as f1 and g1 series workers, are not supported under the Dataflow Service Level Agreement.

Dataflow bills by the number of vCPUs and GB of memory in workers. Billing is independent of the machine type family. You can specify a machine type for your pipeline by setting the appropriate execution parameter at pipeline creation time.

Caution: Shared core machine types such as f1 and g1 series workers are not supported under Dataflow's Service Level Agreement.

Java

To change the machine type, set the --workerMachineType option.

Python

To change the machine type, set the --worker_machine_type option.

Note: The Dataflow service currently does not support jobs with only preemptible virtual machines. Instead, if you would like to save processing costs, consider using the FlexRS batch processing mode that uses a combination of preemptible and non-preemptible resources.

Resource quota

The Dataflow service checks to ensure that your Google Cloud project has the Compute Engine resource quota required to run your job, both to start the job and scale to the maximum number of worker instances. Your job will fail to start if there is not enough resource quota available.

If your Dataflow job deploys Compute Engine virtual machines as a Managed Instance Group, you'll need to ensure your project satisfies some additional quota requirements. Specifically, your project will need one of the following types of quota for each concurrent Dataflow job that you want to run:

• One Instance Group per job
• One Managed Instance Group per job
• One Instance Template per job

Caution: Manually changing your Dataflow job's Instance Template or Managed Instance Group is not recommended or supported. Use Dataflow's pipeline configuration options instead.

Dataflow's Horizontal Autoscaling feature is limited by your project's available Compute Engine quota. If your job has sufficient quota when it starts, but another job uses the remainder of your project's available quota, the first job will run but not be able to fully scale.

However, the Dataflow service does not manage quota increases for jobs that exceed the resource quotas in your project. You are responsible for making any necessary requests for additional resource quota, for which you can use the Google Cloud Console.

IP addresses

By default, Dataflow assigns both public and private IP addresses to worker VMs. A public IP address satisfies one of the criteria for internet access, but a public IP address also counts against your quota of external IP addresses.

If your worker VMs don't need access to the public internet, consider using only internal IP addresses, which don't count against your external quota. For more information on configuring IP addresses, see the following resources:

• Public IP parameter
• Internet access for Dataflow

Persistent disk resources

The Dataflow service is currently limited to 15 persistent disks per worker instance when running a streaming job. Each persistent disk is local to an individual Compute Engine virtual machine. Your job may not have more workers than persistent disks; a 1:1 ratio between workers and disks is the minimum resource allotment.

Jobs using Streaming Engine use 30 GB boot disks. Jobs using Dataflow Shuffle use 25 GB boot disks. For jobs that are not using these offerings, the default size of each persistent disk is 250 GB in batch mode and 400 GB in streaming mode.

Locations

By default, the Dataflow service deploys Compute Engine resources in the us-central1-f zone of the us-central1 region. You can override this setting by specifying the --region parameter. If you need to use a specific zone for your resources, use the --zone parameter when you create your pipeline. However, we recommend that you only specify the region, and leave the zone unspecified. This allows the Dataflow service to automatically select the best zone within the region based on the available zone capacity at the time of the job creation request. For more information, see the regional endpoints documentation.

Streaming Engine

Currently, the Dataflow pipeline runner executes the steps of your streaming pipeline entirely on worker virtual machines, consuming worker CPU, memory, and Persistent Disk storage. Dataflow's Streaming Engine moves pipeline execution out of the worker VMs and into the Dataflow service backend.

Note: Streaming Engine is enabled by default for pipelines developed against Apache Beam Python SDK version 2.21.0 or later using Python 3.

Benefits of Streaming Engine

The Streaming Engine model has the following benefits:

• A reduction in consumed CPU, memory, and Persistent Disk storage resources on the worker VMs. Streaming Engine works best with smaller worker machine types (n1-standard-2 instead of n1-standard-4) and does not require Persistent Disk beyond a small worker boot disk, leading to less resource and quota consumption.
• More responsive Horizontal Autoscaling in response to variations in incoming data volume. Streaming Engine offers smoother, more granular scaling of workers.
• Improved supportability, since you don’t need to redeploy your pipelines to apply service updates.

Most of the reduction in worker resources comes from offloading the work to the Dataflow service. For that reason, there is a charge associated with the use of Streaming Engine. However, the total bill for Dataflow pipelines using Streaming Engine is expected to be approximately the same compared to the total cost of Dataflow pipelines that do not use this option.

Using Streaming Engine

This feature is available in all regions where Dataflow is supported. To see available locations, read Dataflow locations.

Note: Updating an already-running pipeline to use Streaming Engine is not currently supported.

If your pipeline is already running in production and you would like to use Streaming Engine, you need to stop your pipeline using the Dataflow Drain option. Then, specify the Streaming Engine parameter and rerun your pipeline.

Java

Note: Streaming Engine requires the Apache Beam SDK for Java, version 2.10.0 or later.

To use Streaming Engine for your streaming pipelines, specify the following parameter:

• --enableStreamingEngine if you're using Apache Beam SDK for Java versions 2.11.0 or later.
• --experiments=enable_streaming_engine if you're using Apache Beam SDK for Java version 2.10.0.

If you use Dataflow Streaming Engine for your pipeline, do not specify the --zone parameter. Instead, specify the --region parameter and set the value to one of the regions where Streaming Engine is currently available. Dataflow auto-selects the zone in the region you specified. If you do specify the --zone parameter and set it to a zone outside of the available regions, Dataflow reports an error.

Streaming Engine works best with smaller worker machine types, so we recommend that you set --workerMachineType=n1-standard-2. You can also set --diskSizeGb=30 because Streaming Engine only needs space for the worker boot image and local logs. These values are the default values.

Python

Note: Streaming Engine requires the Apache Beam SDK for Python, version 2.16.0 or higher.

Streaming Engine is enabled by default for new Dataflow streaming pipelines when the following conditions are met:

• Pipelines are developed against Apache Beam Python SDK version 2.21.0 or later using Python 3.
• Customer-managed encryption keys are not used.
• Dataflow workers and the regional endpoint for your Dataflow job are located in the same region.

If you would like to disable Streaming Engine in your Python streaming pipeline, specify the following parameter:

--experiments=disable_streaming_engine

Note: This parameter must be specified everytime you want to disable the Streaming Engine.

If you use Python 2, you must still enable Streaming Engine by specifying the following parameter:

--enable_streaming_engine

Caution: On October 7, 2020, Dataflow will stop supporting pipelines using Python 2. Read more information on the Python 2 support on Google Cloud page.

If you use Dataflow Streaming Engine in your pipeline, do not specify the --zone parameter. Instead, specify the --region parameter and set the value to one of the regions where Streaming Engine is currently available. Dataflow auto-selects the zone in the region you specified. If you do specify the --zone parameter and set it to a zone outside of the available regions, Dataflow reports an error.

Streaming Engine works best with smaller worker machine types, so we recommend that you set --machine_type=n1-standard-2. You can also set --disk_size_gb=30 because Streaming Engine only needs space for the worker boot image and local logs. These values are the default values.

Dataflow Shuffle

Dataflow Shuffle is the base operation behind Dataflow transforms such as GroupByKey, CoGroupByKey, and Combine. The Dataflow Shuffle operation partitions and groups data by key in a scalable, efficient, fault-tolerant manner. Currently, Dataflow uses a shuffle implementation that runs entirely on worker virtual machines and consumes worker CPU, memory, and Persistent Disk storage. The service-based Dataflow Shuffle feature, available for batch pipelines only, moves the shuffle operation out of the worker VMs and into the Dataflow service backend.

Batch jobs use Dataflow Shuffle by default.

Benefits of Dataflow Shuffle

The service-based Dataflow Shuffle has the following benefits:

• Faster execution time of batch pipelines for the majority of pipeline job types.
• A reduction in consumed CPU, memory, and Persistent Disk storage resources on the worker VMs.
• Better Horizontal Autoscaling because VMs no longer hold any shuffle data and can therefore be scaled down earlier.
• Better fault tolerance; an unhealthy VM holding Dataflow Shuffle data will not cause the entire job to fail, as would happen if not using the feature.

Most of the reduction in worker resources comes from offloading the shuffle work to the Dataflow service. For that reason, there is a charge associated with the use of Dataflow Shuffle. However, the total bill for Dataflow pipelines using the service-based Dataflow implementation is expected to be less than or equal to the cost of Dataflow pipelines that do not use this option.

For the majority of pipeline job types, Dataflow Shuffle is expected to execute faster than the shuffle implementation running on worker VMs. However, the execution times might vary from run to run. If you are running a pipeline that has important deadlines, we recommend allocating sufficient buffer time before the deadline. In addition, consider requesting a bigger quota for Shuffle.

Exceptions for using Dataflow Shuffle

Batch jobs use Dataflow Shuffle by default. The boot disk size for each batch job reduces to 25 GB instead of the default 250 GB. For some batch jobs, you might be required to turn off Dataflow Shuffle or to modify the size of the disk. Consider the following:

• A worker VM uses part of the 25 GB of disk space for the operating system, binaries, logs, and containers. Jobs that use a significant amount of disk and exceed the remaining disk capacity may fail when you use Dataflow Shuffle.
• Jobs that use a lot of disk I/O may be slow due to the performance of the small disk. For more information about performance differences between disk sizes, see the Compute Engine Persistent Disk Performance page.

To specify a larger disk size for a Dataflow Shuffle job, you can use the --disk_size_gb parameter.

To turn off Dataflow Shuffle, see the following section.

Using Dataflow Shuffle

This feature is available in all regions where Dataflow is supported. To see available locations, read Dataflow locations. If you use the Dataflow Shuffle, the workers must also be deployed in the same region as the regional endpoint.

Note: Performance differences in the asia-northeast1 (Tokyo) region: We recommend using Dataflow Shuffle with large datasets (greater than 1 TB) when you run pipelines in the asia-northeast1 (Tokyo) region. Using Shuffle with smaller datasets in the asia-northeast1 (Tokyo) region does not give you the same performance advantages as Shuffle in other regions.

Java

Batch jobs use Dataflow Shuffle by default. To opt out of using Dataflow Shuffle, specify the following pipeline option:
--experiments=shuffle_mode=appliance.

If you use Dataflow Shuffle for your pipeline, do not specify the zone pipeline options. Instead, specify the region and set the value to one of the regions where Shuffle is currently available. Dataflow autoselects the zone in the region you specified. If you do specify the zone pipeline option and set it to a zone outside of the available regions, Dataflow reports an error. If you set an incompatible combination of region and zone, your job cannot use Dataflow Shuffle.

Python

Note: To use Dataflow Shuffle, you must have Apache Beam SDK for Python version 2.1.0 or later.

Batch jobs use Dataflow Shuffle by default. To opt out of using Dataflow Shuffle, specify the following pipeline option:
--experiments=shuffle_mode=appliance.

If you use Dataflow Shuffle for your pipeline, do not specify the zone pipeline options. Instead, specify the region and set the value to one of the regions where Shuffle is currently available. Dataflow autoselects the zone in the region you specified. If you do specify the zone pipeline option and set it to a zone outside of the available regions, Dataflow reports an error. If you set an incompatible combination of region and zone, your job cannot use Dataflow Shuffle.

Dataflow Flexible Resource Scheduling

Dataflow FlexRS reduces batch processing costs by using advanced scheduling techniques, the Dataflow Shuffle service, and a combination of preemptible virtual machine (VM) instances and regular VMs. By running preemptible VMs and regular VMs in parallel, Dataflow improves the user experience if Compute Engine stops preemptible VM instances during a system event. FlexRS helps to ensure that the pipeline continues to make progress and that you do not lose previous work when Compute Engine preempts your preemptible VMs. For more information about FlexRS, see Using Flexible Resource Scheduling in Dataflow.

Dataflow Runner v2

The current production Dataflow runner utilizes language-specific workers when running Apache Beam pipelines. To improve scalability, generality, extensibility, and efficiency, Dataflow runner is moving to a more services-based architecture. These changes include a more efficient and portable worker architecture packaged together with the Shuffle Service and Streaming Engine.

Dataflow Runner v2 is supported in the following configurations:

Java

Dataflow Runner v2 is available in Preview for the Apache Beam SDK for Java.

Dataflow Runner v2 requires the Apache Beam SDK for Java version 2.30.0 or later.

To enable Runner v2, run your job with the following flag: --experiments=use_runner_v2.

Python

Dataflow Runner v2 requires the Apache Beam SDK version 2.21.0 or later for Python. Runner v2, is the default for Python batch pipelines (version 2.21.0 or later) that were submitted on or after February 1, 2021, and Python streaming pipelines (version 2.21.0 or later). You do not have to make any changes to your pipeline code to take advantage of this new architecture.

Under certain circumstances, your pipeline might not use Runner v2, although the pipeline runs on a supported SDK version. Some examples are Dataflow template jobs, or jobs where Group by Key is used as a side input. In such cases, you can choose to run the job with the --experiments=use_runner_v2 flag.

Benefits of using Dataflow Runner v2

New features will be available on Dataflow Runner v2 only. In addition, the improved efficiency of the Dataflow Runner v2 architecture could lead to performance improvements in your Dataflow jobs.

While using Dataflow Runner v2, you might notice a reduction in your bill.

Dataflow Runner v2 also allows you to pre-build your Python container, which can improve VM startup times and Horizontal Autoscaling performance. To try out this experimental feature, enable the Cloud Build API on your project and submit your pipeline with the following parameter:
--prebuild_sdk_container_engine=cloud_build.

Once your job is finished or stopped, you can remove the pre-built image from Container Registry. The image URL can be found in the Dataflow monitoring UI in Pipeline options.

Note: The pre-build feature requires the Apache Beam SDK for Python, version 2.25.0 or later.

Dataflow Runner v2 supports multi-language pipelines, a feature that enables your Apache Beam pipeline to use transforms defined in other Apache Beam SDKs. Currently, Dataflow Runner v2 supports using Java transforms from a Python SDK pipeline (Preview).

Using Dataflow Runner v2

Dataflow Runner v2 is available in regions that have Dataflow regional endpoints. While this is rolling out, if you would like to try out Dataflow Runner v2, you can use the following parameter:
--experiments=use_runner_v2

Note: Dataflow Runner v2 requires Streaming Engine for streaming jobs and Dataflow Shuffle for batch jobs.

Debugging Dataflow Runner v2 jobs

To debug jobs using Dataflow Runner v2, you should follow standard debugging steps; however, be aware of the following when using Dataflow Runner v2:

• Dataflow Runner v2 jobs run two types of processes on the worker VM—SDK process and the runner harness process. Depending on the pipeline and VM type, there might be one or more SDK processes, but there is only one runner harness process per VM.
• SDK processes run user code and other language-specific functions, while the runner harness process manages everything else.
• The runner harness process waits for all SDK processes to connect to it before starting to request work from Dataflow.
• Jobs might be delayed if the worker VM downloads and installs dependencies during the SDK process startup. If there are issues in an SDK process, such as starting up or installing libraries, the worker reports its status as unhealthy. If the startup times increase, enable the Cloud Build API on your project and submit your pipeline with the following parameter:
  --prebuild_sdk_container_engine=cloud_build.

Note: The pre-build feature requires the Apache Beam SDK for Python, version 2.25.0 or later.

• Worker VM logs—available through the Logs Explorer or the Dataflow monitoring interface—include logs from the runner harness process as well as logs from the SDK processes.
• To diagnose problems in your user code, examine the worker logs from the SDK processes. If you find any errors in the runner harness logs, please contact Support to file a bug.

Dataflow Shielded VM

You can instruct the Dataflow service to use Shielded VM workers. You can learn more about Shielded VM capabilties at Shielded VM.

Java

To use Shielded VM workers, specify the following parameter --dataflowServiceOptions=enable_secure_boot.

Python

To use Shielded VM workers, specify the following parameter --dataflow_service_options=enable_secure_boot.