This page provides best practices for developing and testing your Dataflow pipeline.
Overview
The way that the code for your pipeline is implemented has a significant influence on how well the pipeline performs in production. To help you create pipeline code that works correctly and efficiently, this document explains the following:
- Pipeline runners to support code execution in the different stages of development and deployment.
- Deployment environments that let you run pipelines during development, testing, preproduction, and production.
- Open source pipeline code and templates that you can use as is, or as the basis for new pipelines to accelerate code development.
- A best-practices approach for testing pipeline code. First, this document provides an overview that includes the scope and relationship of different test types, such as unit tests, integration tests, and end-to-end tests. Second, each type of test is explored in detail, including methods to create and integrate with test data, and which pipeline runners to use for each test.
Pipeline runners
During development and testing, you use different Apache Beam runners to run pipeline code. The Apache Beam SDK provides a Direct Runner for local development and testing. Your release automation tooling can also use the Direct Runner for unit tests and integration tests. For example, you can use the Direct Runner within your continuous integration (CI) pipeline.
Pipelines that are deployed to Dataflow use the Dataflow Runner, which runs your pipeline in production-like environments. Additionally, you can use the Dataflow Runner for ad hoc development testing and for end-to-end pipeline tests.
Although this page focuses on running pipelines built using the Apache Beam Java SDK, Dataflow also supports Apache Beam pipelines that were developed using Python and Go. The Apache Beam Java, Python, and Go SDKs are generally available for Dataflow. SQL developers can also use Apache Beam SQL to create pipelines that use familiar SQL dialects.
Set up a deployment environment
To separate users, data, code, and other resources across different stages of development, create deployment environments. When possible, to provide isolated environments for the different stages of pipeline development, use separate Google Cloud projects.
The following sections describe a typical set of deployment environments.
Local environment
The local environment is a developer's workstation. For development and rapid testing, use the Direct Runner to run pipeline code locally.
Pipelines that are run locally using the Direct Runner can interact with remote Google Cloud resources, such as Pub/Sub topics or BigQuery tables. Give individual developers separate Google Cloud projects so that they have a sandbox for ad hoc testing with Google Cloud services.
Some Google Cloud services, such as Pub/Sub and Bigtable, provide emulators for local development. You can use these emulators with the Direct Runner to enable end-to-end local development and testing.
Sandbox environment
The sandbox environment is a Google Cloud project that provides developers with access to Google Cloud services during code development. Pipeline developers can share a Google Cloud project with other developers, or use their own individual projects. Using individual projects reduces planning complexity relating to shared resource usage and quota management.
Developers use the sandbox environment to perform ad hoc pipeline execution with the Dataflow Runner. The sandbox environment is useful for debugging and testing code against a production runner during the code development phase. For example, ad hoc pipeline execution lets developers do the following:
- Observe the effect of code changes on scaling behavior.
- Understand potential differences between the behavior of the Direct Runner and the Dataflow Runner.
- Understand how Dataflow applies graph optimizations.
For ad hoc testing, developers can deploy code from their local environment in order to run Dataflow within their sandbox environment.
Preproduction environment
The preproduction environment is for development phases that need to run in production-like conditions, such as end-to-end testing. Use a separate project for the preproduction environment and configure it to be as similar to production as possible. Similarly, to allow end-to-end tests with production-like scale, make Google Cloud project quotas for Dataflow and other services as similar as possible to the production environment.
Depending on your requirements, you can further separate preproduction into multiple environments. For example, a quality control environment can support the work of quality analysts to test service level objectives (SLOs) such as data correctness, freshness, and performance under different workload conditions.
End-to-end tests include integration with data sources and sinks within the scope of testing. Consider how to make these available within the preproduction environment. You can store test data in the preproduction environment itself. For example, test data is stored in a Cloud Storage bucket with your input data. In other cases, test data might originate from outside the preproduction environment, such as a Pub/Sub topic through a separate subscription that's in the production environment. For streaming pipelines, you can also run end-to-end tests using generated data, for example, using the Dataflow Streaming Data Generator to emulate production-like data characteristics and volumes.
For streaming pipelines, use the preproduction environment to test pipeline updates before any changes are made to production. It's important to test and verify update procedures for streaming pipelines, particularly if you need to coordinate multiple steps, such as when running parallel pipelines to avoid downtime.
Production environment
The production environment is a dedicated Google Cloud project. Continuous delivery copies deployment artifacts to the production environment when all end-to-end tests have passed.
Development best practices
See Dataflow pipeline best practices.
Test your pipeline
In software development, unit tests, integration tests, and end-to-end tests are common types of software testing. These testing types are also applicable to data pipelines.
The Apache Beam SDK provides functionality to enable these tests. Ideally, each type of test targets a different deployment environment. The following diagram illustrates how unit tests, integration tests, and end-to-end tests apply to different parts of your pipeline and data.
The diagram shows the scope of different tests and how they relate to
transforms (DoFn and PTransform subclasses), pipelines, data sources, and
data sinks.
The following sections describe how various formal software tests apply to data pipelines using Dataflow. As you read through this section, refer back to the diagram to understand how the different types of tests are related.
Data sampling
To observe the data at each step of a Dataflow pipeline, enable data sampling during testing. This lets you view the outputs of transforms, to ensure the output is correct.
Unit tests
Unit tests assess the correct functioning of DoFn subclasses and
composite transforms
(PTransform subclasses) by comparing the output of those transforms with a
verified set of data inputs and outputs. Typically, developers can run these
tests in the local environment. The tests can also run automatically through
unit-test automation using continuous integration (CI) in the build
environment.
The Direct Runner runs unit tests using a smaller subset of reference test data that focuses on testing the business logic of your transforms. The test data must be small enough to fit into local memory on the machine that runs the test.
The Apache Beam SDK provides a JUnit rule called
TestPipeline
for unit-testing individual transforms (DoFn subclasses), composite transforms
(PTransform subclasses), and entire pipelines. You can use TestPipeline on a
Apache Beam pipeline runner such as the Direct Runner or the
Dataflow Runner to apply assertions on the contents of
PCollection objects using
PAssert,
as shown in the following code snippet of a
JUnit test class:
@Rule
public final transient TestPipeline p = TestPipeline.create();
@Test
@Category(NeedsRunner.class)
public void myPipelineTest() throws Exception {
final PCollection<String> pcol = p.apply(...)
PAssert.that(pcol).containsInAnyOrder(...);
p.run();
}
Unit tests for individual transforms
By factoring your code into reusable transforms, for example, as top-level or static nested classes, you can create targeted tests for different parts of your pipeline. Aside from the benefits of testing, reusable transforms enhance code maintainability and reusability by naturally encapsulating the business logic of your pipeline into component parts. In contrast, testing individual parts of your pipeline might be difficult if the pipeline uses anonymous inner classes to implement transforms.
The following Java snippet shows the implementation of transforms as anonymous inner classes, which doesn't easily allow testing.
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options)
PCollection<Integer> output =
p.apply("Read from text", TextIO.Read.from(...))
.apply("Split words", ParDo.of(new DoFn() {
// Untestable anonymous transform 1
}))
.apply("Generate anagrams", ParDo.of(new DoFn() {
// Untestable anonymous transform 2
}))
.apply("Count words", Count.perElement());
Compare the previous example with the following one, where the anonymous inner
classes are refactored into named concrete DoFn subclasses. You can create
individual unit tests for each concrete DoFn subclass that makes up the
end-to-end pipeline.
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options)
PCollection<Integer> output =
p.apply("Read from text", TextIO.Read.from(...))
.apply("Split words", ParDo.of(new SplitIntoWordsFn()))
.apply("Generate anagrams", ParDo.of(new GenerateAnagramsFn()))
.apply("Count words", Count.perElement());
Testing each DoFn subclass is similar to unit testing a batch
pipeline that contains a single transform. Use the Create transform to
create a PCollection object of test data, and then pass it to the DoFn
object. Use PAssert to assert that the contents of the PCollection
object are correct. The following Java code example uses the PAssert class to
check for correct output form.
@Rule
public final transient TestPipeline p = TestPipeline.create();
@Test
@Category(NeedsRunner.class)
public void testGenerateAnagramsFn() {
// Create the test input
PCollection<String> words = p.apply(Create.of("friend"));
// Test a single DoFn using the test input
PCollection<String> anagrams =
words.apply("Generate anagrams",