Deploying Pipelines with Orchestration and CI/CD

What it is

A pipeline is operational when it runs by itself on a schedule or event, recovers from routine failures, and can be changed safely through tested, automated releases. Orchestration tools handle the first two; CI/CD handles the third.

How it works

Job automation and orchestration

Composer runs a permanent environment, so it costs money even when idle. For "a few steps, triggered when a file arrives, lowest cost", Workflows triggered by Eventarc is often the better answer.

Packaging Dataflow jobs as templates

A Flex Template packages the pipeline as a container image in Artifact Registry plus a template spec file in Cloud Storage. Operators, Composer or Workflows can then launch it with parameters and no development environment.

gcloud dataflow flex-template run "orders-$(date +%Y%m%d-%H%M)" \
  --template-file-gcs-location=gs://prod-templates/orders.json \
  --region=europe-west1 \
  --parameters=input=gs://landing/orders/*.csv

CI/CD for data pipelines

import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that, equal_to
from orders_pipeline import ParseOrder

def test_parse_order():
    with TestPipeline() as p:
        out = (p | beam.Create(["o1,10.5,uk"])
                 | beam.ParDo(ParseOrder()).with_outputs("bad", main="good"))
        assert_that(out.good, equal_to(
            [{"order_id": "o1", "amount": 10.5, "country": "UK"}]))

Composer DAGs deploy by copying the file into the environment's dags/ folder, which CI can do with one command.

gcloud composer environments storage dags import \
  --environment=prod-composer --location=europe-west1 --source=dags/daily_sales.py

Updating streaming pipelines safely

A drained Pub/Sub pipeline leaves unread messages in the subscription, so the new job picks up where the old one stopped.

A worked example

A team edits a streaming Dataflow job by hand in production, and releases often break it. The engineer moves the code to Git with unit tests. Cloud Build tests every commit, builds a Flex Template image and deploys to a staging project.

After integration tests pass, the pipeline promotes the template to production and updates the running job in place. If the update fails the compatibility check, it drains the old job and starts the new one.

How the exam asks about this

• "A nightly process has 40 interdependent tasks across BigQuery, Dataproc and Dataflow and needs backfills." Cloud Composer.
• "When a file lands, call two APIs and start a load, with the least cost and no servers." Workflows triggered by an Eventarc event.
• "Deploy a new version of a streaming pipeline without losing in-flight data, and the change is not update-compatible." Drain the running job, then start the new one.