You set up and run a workflow by:

1. Creating a workflow template
2. Configuring a managed (ephemeral) cluster or selecting an existing cluster
3. Adding jobs
4. Instantiating the template to run the workflow

> [!IMPORTANT]
> You can [parameterize a workflow template](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/workflow-parameters.md.txt) to use it dynamically for different workflows. You can also [use YAML files](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/using-yamls) or call the [InstantiateInline](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/instantiateInline.md.txt) API to define and run an [inline](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/overview.md.txt#inline) workflow that does not create or modify workflow template resources.

## Create a template

### Google Cloud console

Select **Create Workflow Template** from the Managed Service for Apache Spark
[**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances)
page in Google Cloud console, then confirm or provide template, cluster,
job, and dependencies settings.

### gcloud CLI

Run the following `command` to create a Managed Service for Apache Spark workflow template resource.

```
gcloud dataproc workflow-templates create TEMPLATE_ID \
    --region=REGION
```

Notes:

- <var translate="no">REGION</var>: Specify the [region](https://docs.cloud.google.com/compute/docs/regions-zones.md.txt#available) where your template will run.
- <var translate="no">TEMPLATE_ID</var>: Provide an ID for your template, such as, "workflow-template-1".
- CMEK encryption. You can add the [--kms-key](https://docs.cloud.google.com/sdk/gcloud/reference/dataproc/workflow-templates/create#--kms-key) flag to use [CMEK encryption](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/customer-managed-encryption.md.txt#use_cmek_with_workflow_template_data) on workflow template job arguments.

### REST API

Submit a
[WorkflowTemplate](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates.md.txt#resource-workflowtemplate) as part
of a
[workflowTemplates.create](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/create.md.txt)
request. You can add the
[WorkflowTemplate.EncryptionConfig.kmsKey](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.locations.workflowTemplates.md.txt#encryptionconfig) field to use
[CMEK encryption](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/customer-managed-encryption.md.txt#use_cmek_with_workflow_template_data)
on workflow template job arguments.

### Configure or select a cluster

Managed Service for Apache Spark can create and use a new, "managed" cluster for your
workflow or an existing cluster.

- **Existing cluster:** See
  [Using cluster selectors with workflows](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/cluster-selectors.md.txt)
  to select an existing cluster for your workflow.

- **Managed cluster:** You must configure a managed cluster for
  your workflow. Managed Service for Apache Spark will create this new cluster to run
  workflow jobs, then delete the cluster at the end of the workflow.

  You can configure a managed cluster for your workflow using the Google Cloud console,
  Google Cloud CLI, or the Dataproc API.

  ### Google Cloud console

  Select **Create Workflow Template** from the Managed Service for Apache Spark
  [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances)
  page in Google Cloud console, then confirm or provide template, cluster,
  job, and dependencies settings.

  ### Google Cloud CLI

  Use flags inherited from
  [gcloud dataproc cluster create](https://docs.cloud.google.com/sdk/gcloud/reference/dataproc/clusters/create.md.txt)
  to configure the managed cluster, such as the number of workers and
  the master and worker machine type.

  ```
  gcloud dataproc workflow-templates set-managed-cluster TEMPLATE_ID \
      --cluster-name=CLUSTER_NAME \
      --region=REGION \
      --master-machine-types="type=MACHINE_TYPE_1,rank=0" \
      --master-machine-types="type=MACHINE_TYPE_2,rank=1" \
      --worker-machine-types="type=PRIMARY_WORKER_MACHINE_TYPE_1,rank=0" \
      --worker-machine-types="type=PRIMARY_WORKER_MACHINE_TYPE_2,rank=1" \
      --num-workers=NUMBER \
      --service-account=SERVICE_ACCOUNT
  ```

  Notes:
  - **CLUSTER_NAME:** Managed Service for Apache Spark adds a suffix to the cluster name to ensure uniqueness.
  - **MACHINE_TYPE:** You can use [flexible VM](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/flexible-vms.md.txt) ranked machine types to provide resiliency against resource unavailability.
  - **SERVICE_ACCOUNT:** You can specify a [VM service account](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/service-accounts.md.txt#VM_service_account) for the managed cluster.

  ### REST API

  See
  [WorkflowTemplatePlacement.ManagedCluster](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.locations.workflowTemplates.md.txt#managedcluster),
  which you can provide as part of a completed
  [WorkflowTemplate](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates.md.txt#resource-workflowtemplate)
  submitted with a
  [workflowTemplates.create](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/create.md.txt)
  or
  [workflowTemplates.update](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/update.md.txt)
  request.

  You can use the
  [`GceClusterConfig.serviceAccount`](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/ClusterConfig.md.txt#gceclusterconfig)
  field to specify a [VM service account](https://docs.cloud.google.com/managed-spark/docs/concepts/configuring-clusters/service-accounts.md.txt#VM_service_account)
  for the managed cluster.

  <br />

## Add jobs to a template

All jobs run concurrently unless you specify one or more job dependencies. A
job's dependencies are expressed as a list of other jobs that must finish
successfully before the ultimate job can start. You must provide a `step-id`
for each job. The ID must be unique within the workflow, but does not need to be
unique globally.

### gcloud CLI

Use job type and flags inherited from
[gcloud dataproc jobs submit](https://docs.cloud.google.com/sdk/gcloud/reference/dataproc/jobs/submit)
to define the job to add to the template. You can optionally use the
`‑‑start-after job-id of another workflow job`
flag to have the job start after the completion of one or more other jobs
in the workflow.
The `--max-failures-per-hour` and `--max-failures-total` [restartable job flags](https://docs.cloud.google.com/managed-spark/docs/concepts/jobs/restartable-jobs.md.txt#creating_and_using_restartable_jobs) are not supported in Managed Service for Apache Spark workflow template jobs.

**Examples:**

Add Hadoop job "foo" to the "my-workflow" template.

```
gcloud dataproc workflow-templates add-job hadoop \
    --region=REGION \
    --step-id=foo \
    --workflow-template=my-workflow \
    -- space separated job args
```

Add job "bar" to the "my-workflow" template, which will be run
after workflow job "foo" has completed successfully.

```
gcloud dataproc workflow-templates add-job JOB_TYPE \
    --region=REGION \
    --step-id=bar \
    --start-after=foo \
    --workflow-template=my-workflow \
    -- space separated job args
```

Add another job "baz" to "my-workflow" template to be run after the successful
completion of both "foo" and "bar" jobs.

```
gcloud dataproc workflow-templates add-job JOB_TYPE \
    --region=REGION \
    --step-id=baz \
    --start-after=foo,bar \
    --workflow-template=my-workflow \
    -- space separated job args
```

### REST API

See
[WorkflowTemplate.OrderedJob](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.locations.workflowTemplates.md.txt#orderedjob). This field is provided as part of a
completed
[WorkflowTemplate](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates.md.txt#resource-workflowtemplate)
submitted with a
[workflowTemplates.create](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/create.md.txt)
or
[workflowTemplates.update](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/update.md.txt)
request.
The `maxFailuresPerHour` and `maxFailuresTotal` [OrderedJob.JobScheduling](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/JobScheduling.md.txt) fields are not supported in Managed Service for Apache Spark workflow template jobs.

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.
The `Max restarts per hour` [restartable job option](https://docs.cloud.google.com/managed-spark/docs/concepts/jobs/restartable-jobs.md.txt#creating_and_using_restartable_jobs) is not supported in Managed Service for Apache Spark workflow template jobs.

## Run a workflow

The instantiation of a workflow template runs the workflow defined by the
template. Multiple instantiations of a template are supported---you
can run a workflow multiple times.

### gcloud command

```
gcloud dataproc workflow-templates instantiate TEMPLATE_ID \
    --region=REGION
```

The command returns an operation ID, which you can use to track workflow status.
**Example command and output:**

```
gcloud beta dataproc workflow-templates instantiate my-template-id \
    --region=us-central1
...
WorkflowTemplate [my-template-id] RUNNING
...
Created cluster: my-template-id-rg544az7mpbfa.
Job ID teragen-rg544az7mpbfa RUNNING
Job ID teragen-rg544az7mpbfa COMPLETED
Job ID terasort-rg544az7mpbfa RUNNING
Job ID terasort-rg544az7mpbfa COMPLETED
Job ID teravalidate-rg544az7mpbfa RUNNING
Job ID teravalidate-rg544az7mpbfa COMPLETED
...
Deleted cluster: my-template-id-rg544az7mpbfa.
WorkflowTemplate [my-template-id] DONE

```

### REST API

See [workflowTemplates.instantiate](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/instantiate.md.txt).

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.

## Workflow job failures

A failure in any job in a workflow will cause the workflow to fail.
Managed Service for Apache Spark will seek to mitigate the effect of failures by causing all
concurrently executing jobs to fail and preventing subsequent jobs
from starting.

## Monitor and list a workflow

### gcloud CLI

To monitor a workflow:

```
gcloud dataproc operations describe OPERATION_ID \
    --region=REGION
```

**Note:** The operation-id is returned when you instantiate the workflow
with `gcloud dataproc workflow-templates instantiate` (see
[Run a workflow](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/use-workflows.md.txt#run_a_workflow)).

To list workflow status:

```
gcloud dataproc operations list \
    --region=REGION \
    --filter="labels.goog-dataproc-operation-type=WORKFLOW AND status.state=RUNNING"
```

### REST API

To monitor a workflow, use the Managed Service for Apache Spark
[operations.get](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.operations/get.md.txt)
API.

To list running workflows, use the Managed Service for Apache Spark
[operations.list](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.operations/list.md.txt)
API with a label filter.

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.

## Terminate a workflow

You can end a workflow using the Google Cloud CLI or by calling
the Dataproc API.

> [!NOTE]
> **Note:** Ending a workflow cancels running workflow jobs and, if the workflow runs on a managed (ephemeral) cluster, deletes the managed cluster.

### gcloud command

```
gcloud dataproc operations cancel OPERATION_ID \
    --region=REGION
```
**Note:** The operation-id that is returned when you instantiate the workflow with `gcloud dataproc workflow-templates instantiate` (see [Run a workflow](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/use-workflows.md.txt#run_a_workflow)).

### REST API

See the [operations.cancel](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.operations/cancel.md.txt)
API.

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.

## Update a workflow template

Updates don't affect running workflows. The new template version will only
apply to new workflows.

### gcloud CLI

Workflow templates can be updated by issuing new `gcloud workflow-templates`
commands that reference an existing workflow template-id:

- [`gcloud dataproc workflow-templates add job`](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/use-workflows.md.txt#add_jobs_to_a_template)
- [`gcloud dataproc workflow-templates set-managed-cluster`](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/use-workflows.md.txt#configure_or_select_a_cluster)
- [`gcloud dataproc workflow-templates set-cluster-selector`](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/cluster-selectors.md.txt#add_a_cluster_selector_to_a_template)
to an existing workflow template.

### REST API

To make an update to a template with the REST API:

1. Call [workflowTemplates.get](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/get.md.txt), which returns the current template with the `version` field filled in with the current server version.
2. Make updates to the fetched template.
3. Call [workflowTemplates.update](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.locations.workflowTemplates/update.md.txt) with the updated template.

As a guard against concurrent modifications, a request to update a template must specify the current server version in the `workflowTemplate.version` field.

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.

## Delete a workflow template

### gcloud CLI

```
gcloud dataproc workflow-templates delete TEMPLATE_ID \
    --region=REGION
```

**Note:** The operation-id that is returned when you instantiate the workflow
with `gcloud dataproc workflow-templates instantiate` (see
[Run a workflow](https://docs.cloud.google.com/managed-spark/docs/concepts/workflows/use-workflows.md.txt#run_a_workflow)).

### REST API

See [workflowTemplates.delete](https://docs.cloud.google.com/managed-spark/docs/reference/rest/v1/projects.regions.workflowTemplates/delete.md.txt).

### Console

You can view existing workflow templates and instantiated workflows from
the Managed Service for Apache Spark [**Workflows**](https://console.cloud.google.com/dataproc/workflows/instances) page in Google Cloud console.