Deploy Google Cloud Dataflow Jobs

The gcp:dataflow/job:Job resource, part of the Pulumi GCP provider, launches and manages Dataflow jobs that execute Apache Beam pipelines on Google Compute Engine. This guide focuses on three capabilities: template-based deployment, Streaming Engine configuration, and job lifecycle management.

Dataflow jobs require pre-built templates stored in GCS, temporary storage buckets for intermediate data, and may reference Pub/Sub topics or subscriptions for streaming workloads. The examples are intentionally small. Combine them with your own templates, storage infrastructure, and data sources.

Launch a batch job from a template

Most deployments start by launching a job from a pre-built template, which packages pipeline logic and dependencies for reuse without rebuilding code.

import * as pulumi from "@pulumi/pulumi";
import * as gcp from "@pulumi/gcp";

const bigDataJob = new gcp.dataflow.Job("big_data_job", {
    name: "dataflow-job",
    templateGcsPath: "gs://my-bucket/templates/template_file",
    tempGcsLocation: "gs://my-bucket/tmp_dir",
    parameters: {
        foo: "bar",
        baz: "qux",
    },
});

import pulumi
import pulumi_gcp as gcp

big_data_job = gcp.dataflow.Job("big_data_job",
    name="dataflow-job",
    template_gcs_path="gs://my-bucket/templates/template_file",
    temp_gcs_location="gs://my-bucket/tmp_dir",
    parameters={
        "foo": "bar",
        "baz": "qux",
    })

package main

import (
	"github.com/pulumi/pulumi-gcp/sdk/v9/go/gcp/dataflow"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		_, err := dataflow.NewJob(ctx, "big_data_job", &dataflow.JobArgs{
			Name:            pulumi.String("dataflow-job"),
			TemplateGcsPath: pulumi.String("gs://my-bucket/templates/template_file"),
			TempGcsLocation: pulumi.String("gs://my-bucket/tmp_dir"),
			Parameters: pulumi.StringMap{
				"foo": pulumi.String("bar"),
				"baz": pulumi.String("qux"),
			},
		})
		if err != nil {
			return err
		}
		return nil
	})
}

using System.Collections.Generic;
using System.Linq;
using Pulumi;
using Gcp = Pulumi.Gcp;

return await Deployment.RunAsync(() => 
{
    var bigDataJob = new Gcp.Dataflow.Job("big_data_job", new()
    {
        Name = "dataflow-job",
        TemplateGcsPath = "gs://my-bucket/templates/template_file",
        TempGcsLocation = "gs://my-bucket/tmp_dir",
        Parameters = 
        {
            { "foo", "bar" },
            { "baz", "qux" },
        },
    });

});

package generated_program;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.core.Output;
import com.pulumi.gcp.dataflow.Job;
import com.pulumi.gcp.dataflow.JobArgs;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var bigDataJob = new Job("bigDataJob", JobArgs.builder()
            .name("dataflow-job")
            .templateGcsPath("gs://my-bucket/templates/template_file")
            .tempGcsLocation("gs://my-bucket/tmp_dir")
            .parameters(Map.ofEntries(
                Map.entry("foo", "bar"),
                Map.entry("baz", "qux")
            ))
            .build());

    }
}

resources:
  bigDataJob:
    type: gcp:dataflow:Job
    name: big_data_job
    properties:
      name: dataflow-job
      templateGcsPath: gs://my-bucket/templates/template_file
      tempGcsLocation: gs://my-bucket/tmp_dir
      parameters:
        foo: bar
        baz: qux

The templateGcsPath points to your compiled pipeline template in GCS. The tempGcsLocation specifies where Dataflow writes intermediate data during execution. The parameters object passes key-value pairs to your pipeline’s runtime options, allowing you to customize behavior without modifying the template.

Process streaming data with Streaming Engine

Streaming pipelines continuously process data from sources like Pub/Sub, writing results to storage or other services. Streaming Engine offloads state management from workers to Google’s backend.

import * as pulumi from "@pulumi/pulumi";
import * as gcp from "@pulumi/gcp";

const topic = new gcp.pubsub.Topic("topic", {name: "dataflow-job1"});
const bucket1 = new gcp.storage.Bucket("bucket1", {
    name: "tf-test-bucket1",
    location: "US",
    forceDestroy: true,
});
const bucket2 = new gcp.storage.Bucket("bucket2", {
    name: "tf-test-bucket2",
    location: "US",
    forceDestroy: true,
});
const pubsubStream = new gcp.dataflow.Job("pubsub_stream", {
    name: "tf-test-dataflow-job1",
    templateGcsPath: "gs://my-bucket/templates/template_file",
    tempGcsLocation: "gs://my-bucket/tmp_dir",
    enableStreamingEngine: true,
    parameters: {
        inputFilePattern: pulumi.interpolate`${bucket1.url}/*.json`,
        outputTopic: topic.id,
    },
    transformNameMapping: {
        name: "test_job",
        env: "test",
    },
    onDelete: "cancel",
});

import pulumi
import pulumi_gcp as gcp

topic = gcp.pubsub.Topic("topic", name="dataflow-job1")
bucket1 = gcp.storage.Bucket("bucket1",
    name="tf-test-bucket1",
    location="US",
    force_destroy=True)
bucket2 = gcp.storage.Bucket("bucket2",
    name="tf-test-bucket2",
    location="US",
    force_destroy=True)
pubsub_stream = gcp.dataflow.Job("pubsub_stream",
    name="tf-test-dataflow-job1",
    template_gcs_path="gs://my-bucket/templates/template_file",
    temp_gcs_location="gs://my-bucket/tmp_dir",
    enable_streaming_engine=True,
    parameters={
        "inputFilePattern": bucket1.url.apply(lambda url: f"{url}/*.json"),
        "outputTopic": topic.id,
    },
    transform_name_mapping={
        "name": "test_job",
        "env": "test",
    },
    on_delete="cancel")

package main

import (
	"fmt"

	"github.com/pulumi/pulumi-gcp/sdk/v9/go/gcp/dataflow"
	"github.com/pulumi/pulumi-gcp/sdk/v9/go/gcp/pubsub"
	"github.com/pulumi/pulumi-gcp/sdk/v9/go/gcp/storage"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		topic, err := pubsub.NewTopic(ctx, "topic", &pubsub.TopicArgs{
			Name: pulumi.String("dataflow-job1"),
		})
		if err != nil {
			return err
		}
		bucket1, err := storage.NewBucket(ctx, "bucket1", &storage.BucketArgs{
			Name:         pulumi.String("tf-test-bucket1"),
			Location:     pulumi.String("US"),
			ForceDestroy: pulumi.Bool(true),
		})
		if err != nil {
			return err
		}
		_, err = storage.NewBucket(ctx, "bucket2", &storage.BucketArgs{
			Name:         pulumi.String("tf-test-bucket2"),
			Location:     pulumi.String("US"),
			ForceDestroy: pulumi.Bool(true),
		})
		if err != nil {
			return err
		}
		_, err = dataflow.NewJob(ctx, "pubsub_stream", &dataflow.JobArgs{
			Name:                  pulumi.String("tf-test-dataflow-job1"),
			TemplateGcsPath:       pulumi.String("gs://my-bucket/templates/template_file"),
			TempGcsLocation:       pulumi.String("gs://my-bucket/tmp_dir"),
			EnableStreamingEngine: pulumi.Bool(true),
			Parameters: pulumi.StringMap{
				"inputFilePattern": bucket1.Url.ApplyT(func(url string) (string, error) {
					return fmt.Sprintf("%v/*.json", url), nil
				}).(pulumi.StringOutput),
				"outputTopic": topic.ID(),
			},
			TransformNameMapping: pulumi.StringMap{
				"name": pulumi.String("test_job"),
				"env":  pulumi.String("test"),
			},
			OnDelete: pulumi.String("cancel"),
		})
		if err != nil {
			return err
		}
		return nil
	})
}

using System.Collections.Generic;
using System.Linq;
using Pulumi;
using Gcp = Pulumi.Gcp;

return await Deployment.RunAsync(() => 
{
    var topic = new Gcp.PubSub.Topic("topic", new()
    {
        Name = "dataflow-job1",
    });

    var bucket1 = new Gcp.Storage.Bucket("bucket1", new()
    {
        Name = "tf-test-bucket1",
        Location = "US",
        ForceDestroy = true,
    });

    var bucket2 = new Gcp.Storage.Bucket("bucket2", new()
    {
        Name = "tf-test-bucket2",
        Location = "US",
        ForceDestroy = true,
    });

    var pubsubStream = new Gcp.Dataflow.Job("pubsub_stream", new()
    {
        Name = "tf-test-dataflow-job1",
        TemplateGcsPath = "gs://my-bucket/templates/template_file",
        TempGcsLocation = "gs://my-bucket/tmp_dir",
        EnableStreamingEngine = true,
        Parameters = 
        {
            { "inputFilePattern", bucket1.Url.Apply(url => $"{url}/*.json") },
            { "outputTopic", topic.Id },
        },
        TransformNameMapping = 
        {
            { "name", "test_job" },
            { "env", "test" },
        },
        OnDelete = "cancel",
    });

});

package generated_program;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.core.Output;
import com.pulumi.gcp.pubsub.Topic;
import com.pulumi.gcp.pubsub.TopicArgs;
import com.pulumi.gcp.storage.Bucket;
import com.pulumi.gcp.storage.BucketArgs;
import com.pulumi.gcp.dataflow.Job;
import com.pulumi.gcp.dataflow.JobArgs;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var topic = new Topic("topic", TopicArgs.builder()
            .name("dataflow-job1")
            .build());

        var bucket1 = new Bucket("bucket1", BucketArgs.builder()
            .name("tf-test-bucket1")
            .location("US")
            .forceDestroy(true)
            .build());

        var bucket2 = new Bucket("bucket2", BucketArgs.builder()
            .name("tf-test-bucket2")
            .location("US")
            .forceDestroy(true)
            .build());

        var pubsubStream = new Job("pubsubStream", JobArgs.builder()
            .name("tf-test-dataflow-job1")
            .templateGcsPath("gs://my-bucket/templates/template_file")
            .tempGcsLocation("gs://my-bucket/tmp_dir")
            .enableStreamingEngine(true)
            .parameters(Map.ofEntries(
                Map.entry("inputFilePattern", bucket1.url().applyValue(_url -> String.format("%s/*.json", _url))),
                Map.entry("outputTopic", topic.id())
            ))
            .transformNameMapping(Map.ofEntries(
                Map.entry("name", "test_job"),
                Map.entry("env", "test")
            ))
            .onDelete("cancel")
            .build());

    }
}

resources:
  topic:
    type: gcp:pubsub:Topic
    properties:
      name: dataflow-job1
  bucket1:
    type: gcp:storage:Bucket
    properties:
      name: tf-test-bucket1
      location: US
      forceDestroy: true
  bucket2:
    type: gcp:storage:Bucket
    properties:
      name: tf-test-bucket2
      location: US
      forceDestroy: true
  pubsubStream:
    type: gcp:dataflow:Job
    name: pubsub_stream
    properties:
      name: tf-test-dataflow-job1
      templateGcsPath: gs://my-bucket/templates/template_file
      tempGcsLocation: gs://my-bucket/tmp_dir
      enableStreamingEngine: true
      parameters:
        inputFilePattern: ${bucket1.url}/*.json
        outputTopic: ${topic.id}
      transformNameMapping:
        name: test_job
        env: test
      onDelete: cancel

When enableStreamingEngine is true, Dataflow manages pipeline state in its backend rather than on worker VMs, improving autoscaling and reducing worker resource requirements. The onDelete property controls termination behavior: “cancel” stops immediately, while “drain” (the default) processes in-flight data before stopping. The transformNameMapping property supports pipeline updates by mapping old transform names to new ones.

Handle job lifecycle with unique names

Dataflow job names must be unique within a project. When jobs terminate and recreate, name conflicts can block new launches.

import * as pulumi from "@pulumi/pulumi";
import * as gcp from "@pulumi/gcp";
import * as random from "@pulumi/random";

const config = new pulumi.Config();
const bigDataJobSubscriptionId = config.get("bigDataJobSubscriptionId") || "projects/myproject/subscriptions/messages";
const bigDataJobNameSuffix = new random.index.Id("big_data_job_name_suffix", {
    byteLength: 4,
    keepers: {
        region: region,
        subscriptionId: bigDataJobSubscriptionId,
    },
});
const bigDataJob = new gcp.dataflow.FlexTemplateJob("big_data_job", {
    name: `dataflow-flextemplates-job-${bigDataJobNameSuffix.dec}`,
    region: region,
    containerSpecGcsPath: "gs://my-bucket/templates/template.json",
    skipWaitOnJobTermination: true,
    parameters: {
        inputSubscription: bigDataJobSubscriptionId,
    },
});

import pulumi
import pulumi_gcp as gcp
import pulumi_random as random

config = pulumi.Config()
big_data_job_subscription_id = config.get("bigDataJobSubscriptionId")
if big_data_job_subscription_id is None:
    big_data_job_subscription_id = "projects/myproject/subscriptions/messages"
big_data_job_name_suffix = random.index.Id("big_data_job_name_suffix",
    byte_length=4,
    keepers={
        region: region,
        subscriptionId: big_data_job_subscription_id,
    })
big_data_job = gcp.dataflow.FlexTemplateJob("big_data_job",
    name=f"dataflow-flextemplates-job-{big_data_job_name_suffix['dec']}",
    region=region,
    container_spec_gcs_path="gs://my-bucket/templates/template.json",
    skip_wait_on_job_termination=True,
    parameters={
        "inputSubscription": big_data_job_subscription_id,
    })

package main

import (
	"fmt"

	"github.com/pulumi/pulumi-gcp/sdk/v9/go/gcp/dataflow"
	"github.com/pulumi/pulumi-random/sdk/v4/go/random"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi/config"
)

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		cfg := config.New(ctx, "")
		bigDataJobSubscriptionId := "projects/myproject/subscriptions/messages"
		if param := cfg.Get("bigDataJobSubscriptionId"); param != "" {
			bigDataJobSubscriptionId = param
		}
		bigDataJobNameSuffix, err := random.NewId(ctx, "big_data_job_name_suffix", &random.IdArgs{
			ByteLength: 4,
			Keepers: map[string]interface{}{
				"region":         region,
				"subscriptionId": bigDataJobSubscriptionId,
			},
		})
		if err != nil {
			return err
		}
		_, err = dataflow.NewFlexTemplateJob(ctx, "big_data_job", &dataflow.FlexTemplateJobArgs{
			Name:                     pulumi.Sprintf("dataflow-flextemplates-job-%v", bigDataJobNameSuffix.Dec),
			Region:                   pulumi.Any(region),
			ContainerSpecGcsPath:     pulumi.String("gs://my-bucket/templates/template.json"),
			SkipWaitOnJobTermination: pulumi.Bool(true),
			Parameters: pulumi.StringMap{
				"inputSubscription": pulumi.String(bigDataJobSubscriptionId),
			},
		})
		if err != nil {
			return err
		}
		return nil
	})
}

using System.Collections.Generic;
using System.Linq;
using Pulumi;
using Gcp = Pulumi.Gcp;
using Random = Pulumi.Random;

return await Deployment.RunAsync(() => 
{
    var config = new Config();
    var bigDataJobSubscriptionId = config.Get("bigDataJobSubscriptionId") ?? "projects/myproject/subscriptions/messages";
    var bigDataJobNameSuffix = new Random.Index.Id("big_data_job_name_suffix", new()
    {
        ByteLength = 4,
        Keepers = 
        {
            { "region", region },
            { "subscriptionId", bigDataJobSubscriptionId },
        },
    });

    var bigDataJob = new Gcp.Dataflow.FlexTemplateJob("big_data_job", new()
    {
        Name = $"dataflow-flextemplates-job-{bigDataJobNameSuffix.Dec}",
        Region = region,
        ContainerSpecGcsPath = "gs://my-bucket/templates/template.json",
        SkipWaitOnJobTermination = true,
        Parameters = 
        {
            { "inputSubscription", bigDataJobSubscriptionId },
        },
    });

});

package generated_program;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.core.Output;
import com.pulumi.random.Id;
import com.pulumi.random.IdArgs;
import com.pulumi.gcp.dataflow.FlexTemplateJob;
import com.pulumi.gcp.dataflow.FlexTemplateJobArgs;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        final var config = ctx.config();
        final var bigDataJobSubscriptionId = config.get("bigDataJobSubscriptionId").orElse("projects/myproject/subscriptions/messages");
        var bigDataJobNameSuffix = new Id("bigDataJobNameSuffix", IdArgs.builder()
            .byteLength(4)
            .keepers(Map.ofEntries(
                Map.entry("region", region),
                Map.entry("subscriptionId", bigDataJobSubscriptionId)
            ))
            .build());

        var bigDataJob = new FlexTemplateJob("bigDataJob", FlexTemplateJobArgs.builder()
            .name(String.format("dataflow-flextemplates-job-%s", bigDataJobNameSuffix.dec()))
            .region(region)
            .containerSpecGcsPath("gs://my-bucket/templates/template.json")
            .skipWaitOnJobTermination(true)
            .parameters(Map.of("inputSubscription", bigDataJobSubscriptionId))
            .build());

    }
}

configuration:
  bigDataJobSubscriptionId:
    type: string
    default: projects/myproject/subscriptions/messages
resources:
  bigDataJobNameSuffix:
    type: random:Id
    name: big_data_job_name_suffix
    properties:
      byteLength: 4
      keepers:
        region: ${region}
        subscriptionId: ${bigDataJobSubscriptionId}
  bigDataJob:
    type: gcp:dataflow:FlexTemplateJob
    name: big_data_job
    properties:
      name: dataflow-flextemplates-job-${bigDataJobNameSuffix.dec}
      region: ${region}
      containerSpecGcsPath: gs://my-bucket/templates/template.json
      skipWaitOnJobTermination: true
      parameters:
        inputSubscription: ${bigDataJobSubscriptionId}

The random.index.Id resource generates a unique suffix for each job name. The keepers property triggers a new suffix when specified values change, ensuring name uniqueness across job recreations. Setting skipWaitOnJobTermination to true allows Pulumi to proceed without waiting for the job to fully drain or cancel, but requires name uniqueness logic to avoid conflicts on the next launch.

Beyond these examples

These snippets focus on specific job-level features: template-based deployment, streaming engine and continuous processing, and job lifecycle and name management. They’re intentionally minimal rather than full data pipeline deployments.

The examples may reference pre-existing infrastructure such as GCS buckets for templates and temporary data, and Pub/Sub topics and subscriptions. They focus on configuring the job rather than provisioning the surrounding data infrastructure.

To keep things focused, common job patterns are omitted, including:

  • Network configuration (network, subnetwork, ipConfiguration)
  • Worker tuning (machineType, maxWorkers)
  • Security (kmsKeyName, serviceAccountEmail)
  • Job termination behavior (onDelete drain vs cancel)

These omissions are intentional: the goal is to illustrate how each job feature is wired, not provide drop-in pipeline modules. See the Dataflow Job resource reference for all available configuration options.

Let's deploy Google Cloud Dataflow Jobs

Get started with Pulumi Cloud, then follow our quick setup guide to deploy this infrastructure.

Try Pulumi Cloud for FREE

Frequently Asked Questions

Job Lifecycle & Deletion
Why does my Dataflow job keep recreating after it completes or fails?
Dataflow jobs are considered ’existing’ only while in non-terminal states. When a job reaches a terminal state like FAILED, COMPLETE, or CANCELLED, it’s automatically recreated on the next apply. This is expected behavior for continuously-running jobs.
Why is pulumi destroy taking so long for my Dataflow job?
When onDelete is set to "drain" (the default), Pulumi waits for all in-flight data to finish processing before completing the destroy operation. To speed this up, either set onDelete to "cancel" for immediate termination, or use skipWaitOnJobTermination set to true.
What's the difference between drain and cancel when deleting a job?
With "drain", no new data enters the pipeline, but existing data finishes processing before termination. With "cancel", the job terminates immediately and any data already written remains, but no new data is processed.
How do I avoid name conflicts when using skipWaitOnJobTermination?
Use a random.Id resource with keepers to generate unique name suffixes between job instances. For example: name: pulumi.interpolate\dataflow-job-${randomId.dec}`` ensures each job has a unique name.
Configuration & Parameters
What should I put in the parameters field?
Use parameters only for template-specific key/value pairs that get forwarded to your pipeline’s options (keys are case-sensitive based on pipeline language). Don’t configure Dataflow-specific options here.
What format should I use for serviceAccountEmail?
Use just the email address without any prefix, like myserviceaccount@myproject.iam.gserviceaccount.com. Don’t include serviceAccount: or other prefixes.
Why aren't my labels showing up in GCP?
The labels field is non-authoritative and only manages labels present in your configuration. To see all labels on the resource (including those set by other clients), check the effectiveLabels output property.
Immutability & Updates
What properties can't I change after creating a Dataflow job?
These properties are immutable: name, project, region, zone, and maxWorkers. Changing any of these requires recreating the job.
How do I update an existing pipeline?
Use transformNameMapping to map transform name prefixes from the old job to the new job. This field is only used during updates, not initial creation.
Networking & Streaming
What URL format do I need for Shared VPC subnetworks?
For Shared VPC networks, use the complete URL format: googleapis.com/compute/v1/projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME. Short-form paths won’t work.
Is Streaming Engine enabled by default?
Streaming Engine is enabled by default for pipelines developed against the Beam SDK for Python v2.21.0 or later when using Python 3. For other configurations, set enableStreamingEngine to true explicitly.

Using a different cloud?

Explore analytics guides for other cloud providers:

AWS Guides Azure Guides