Updating an existing pipeline

The Apache Beam SDKs provide a way to update an ongoing streaming job on the Dataflow managed service with new pipeline code.

There are various reasons why you might want to update your existing Dataflow job:

  • You want to enhance or otherwise improve your pipeline code.
  • You want to fix bugs in your pipeline code.
  • You want to update your pipeline to handle changes in data format, or to account for version or other changes in your data source.

When you update your job, the Dataflow service performs a compatibility check between your currently-running job and your potential replacement job. The compatibility check ensures that things like intermediate state information and buffered data can be transferred from your prior job to your replacement job.

The update process and its effects

When you update a job on the Dataflow service, you replace the existing job with a new job that runs your updated pipeline code. The Dataflow service retains the job name, but runs the replacement job with an updated Job ID.

The replacement job preserves any intermediate state data from the prior job, as well as any buffered data records or metadata currently "in-flight" from the prior job. For example, some records in your pipeline might be buffered while waiting for a window to resolve.

In-flight data

"In-flight" data will still be processed by the transforms in your new pipeline. However, additional transforms that you add in your replacement pipeline code may or may not take effect, depending on where the records are buffered. For example, let's say your existing pipeline has the following transforms:

Java: SDK 2.x

  p.apply("Read", ReadStrings())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Format' >> FormatStrings()

Java: SDK 1.x

  p.apply(ReadStrings().named("Read"))
   .apply(FormatStrings().named("Format"));

You can replace your job with new pipeline code, as follows:

Java: SDK 2.x

  p.apply("Read", ReadStrings())
   .apply("Remove", RemoveStringsStartingWithA())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Remove', RemoveStringsStartingWithA()
    | 'Format' >> FormatStrings()

Java: SDK 1.x

  p.apply(ReadStrings().named("Read"))
   .apply(RemoveStringsStartingWithA().named("Remove"))
   .apply(FormatStrings().named("Format"));

Even though you've added a transform to filter out strings that begin with the letter "A", the next transform (FormatStrings) may still see buffered or in-flight strings that begin with "A" that were transferred over from the prior job.

Changing windowing

You can change windowing and trigger strategies for the PCollections in your replacement pipeline, but use caution. Changing the windowing or trigger strategies will not affect data that is already buffered or otherwise in-flight.

We recommend that you attempt only smaller changes to your pipeline's windowing, such as changing the duration of fixed- or sliding-time windows. Making major changes to windowing or triggers, like changing the windowing algorithm, might have unpredictable results on your pipeline output.

Launching your replacement job

To update your job, you'll need to launch a new job to replace the ongoing job. When you launch your replacement job, you'll need to set the following pipeline options to perform the update process in addition to the job's regular options:

Java

  • Pass the --update option.
  • Set the --jobName option in PipelineOptions to the same name as the job you want to update.
  • If any transform names in your pipeline have changed, you must supply a transform mapping and pass it using the --transformNameMapping option.

Python

  • Pass the --update option.
  • Set the --job_name option in PipelineOptions to the same name as the job you want to update.
  • If any transform names in your pipeline have changed, you must supply a transform mapping and pass it using the --transform_name_mapping option.

Specifying your replacement job name

Java

When you launch your replacement job, the value you pass for the --jobName option must match exactly the name of the job you want to replace.

Python

When you launch your replacement job, the value you pass for the --job_name option must match exactly the name of the job you want to replace.

To find the correct job name value, select your prior job in the Dataflow Monitoring Interface and find the Job Name field in the Summary tab:

Figure 1: The Summary tab for a running Dataflow job, with the Job Name field highlighted.

Alternatively, you can query a list of existing jobs by using the Dataflow Command-line Interface. Enter the command gcloud beta dataflow jobs list into your shell or terminal window to obtain a list of Dataflow jobs in your Google Cloud project, and find the NAME field for the job you want to replace:

ID                                        NAME                                 TYPE       CREATION_TIME        STATUS
2015-07-28_17_02_27-7257409117866690674   windowedwordcount-johndoe-0729000214 Streaming  2015-07-28 17:02:28  Running

Creating the transform mapping

Java

If your replacement pipeline has changed any transform names from those in your prior pipeline, the Dataflow service requires a transform mapping. The transform mapping maps the named transforms in your prior pipeline code to names in your replacement pipeline code. You can pass the mapping by using the --transformNameMapping command-line option, using the following general format:

--transformNameMapping= . 
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

You only need to provide mapping entries in --transformNameMapping for transform names that have changed between your prior pipeline and your replacement pipeline.

Python

If your replacement pipeline has changed any transform names from those in your prior pipeline, the Dataflow service requires a transform mapping. The transform mapping maps the named transforms in your prior pipeline code to names in your replacement pipeline code. You can pass the mapping by using the --transform_name_mapping command-line option, using the following general format:

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

You only need to provide mapping entries in --transform_name_mapping for transform names that have changed between your prior pipeline and your replacement pipeline.

Determining transform names

The transform name in each instance in the map is the name you supplied when you applied the transform in your pipeline. For example:

Java: SDK 2.x

.apply("FormatResults", ParDo
  .of(new DoFn<KV<String, Long>>, String>() {
    ...
   }
}))

Python

| 'FormatResults' >> beam.ParDo(MyDoFn())

Java: SDK 1.x

.apply(ParDo
  .named("FormatResults")
  .of(new DoFn<KV<String, Long>>, String>() {
    ...
   }
}))

You can also obtain the transform names for your prior job by examining the job's execution graph in the Dataflow Monitoring Interface:

Figure 2: The execution graph for a WordCount pipeline as shown in the Dataflow Monitoring Interface.

Naming in composite transforms

Transform names are hierarchical, based on the transform hierarchy in your pipeline. If your pipeline has a composite transform, the nested transforms are named in terms of their containing transform. For example, suppose your pipeline contains a composite transform named CountWidgets, which contains an inner transform named Parse. The inner transform's full name will be CountWidgets/Parse, and you must specify that full name in your transform mapping.

If your new pipeline maps a composite transform to a different name, all of the nested transforms are also automatically renamed. You'll need to specify the changed names for the inner transforms in your transform mapping.

Refactoring the transform hierarchy

If your replacement pipeline uses a different transform hierarchy than your prior pipeline (e.g. because you refactored your composite transforms, or your pipeline depends on a composite transform from a library that changed), you'll need to explicitly declare the mapping.

For example, let's suppose your prior pipeline applied a composite transform, CountWidgets, that contained an inner transform named Parse. Now, let's say your replacement pipeline refactors CountWidgets, and nests Parse inside another transform named Scan. For your update to succeed, you must explicitly map the prior pipeline's complete transform name (CountWidgets/Parse) to the new pipeline's transform name (CountWidgets/Scan/Parse):

Java

--transformNameMapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

If you delete a transform entirely in your replacement pipeline, you must provide a null mapping. Suppose our replacement pipeline removes the CountWidgets/Parse transform entirely:

--transformNameMapping={"CountWidgets/Parse":""}

Python

--transform_name_mapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

If you delete a transform entirely in your replacement pipeline, you must provide a null mapping. Suppose our replacement pipeline removes the CountWidgets/Parse transform entirely:

--transform_name_mapping={"CountWidgets/Parse":""}