Automating Dataform Workflows: How to Trigger Pipelines from GCS File Events

As data engineers, we often need to trigger data transformation pipelines immediately after new data lands in our storage buckets. While there are many orchestration tools available, sometimes a lightweight, event-driven approach is the perfect fit.

In this post, I’ll share a solution for automatically initiating a Dataform workflow whenever a file event (like a new upload) is detected in a Google Cloud Storage (GCS) bucket. This architecture leverages Eventarc and Cloud Functions (2nd Gen) to bridge the gap between GCS and the Dataform API.

The Operational Workflow

The goal is simple: a file drops into a bucket, and our Dataform pipeline starts running. Here is the step-by-step flow of how this solution works:

1. File Event: A file is uploaded or modified in a designated GCS bucket.
2. Event Detection: Eventarc detects this specific file event.
3. Trigger: Eventarc invokes a Cloud Function, passing along the necessary event metadata.
4. Execution: The Cloud Function authenticates and communicates with the Dataform API to create a new workflow invocation.

IAM Permissions: The Key to Security

One of the trickiest parts of setting up event-driven architectures is getting the Identity and Access Management (IAM) right. You need to ensure each component has exactly the permissions it needs — no more, no less. While you can configure IAM to align with your specific security requirements, the following represents one possible sample architecture.

Here is the breakdown of the service accounts (SA) and roles required for this setup:

1. Cloud Function & Eventarc Identity

This service account (often the default compute SA) acts as the runtime identity for the Cloud Function and the trigger identity for Eventarc.

Roles Required:

• Dataform Editor: Allows the function to call Dataform APIs (e.g., list compilation results, create workflow invocations).
• Eventarc Event Receiver: Required to receive events from the Eventarc trigger.
• Cloud Run Invoker: Allows the service account to invoke the underlying Cloud Run service of the 2nd Gen Cloud Function.
• Service Account User: Needed on the Dataform Workflow Executor SA (see below). This allows the function to tell Dataform: “Run this workflow using that specific service account.”

2. Dataform Workflow Executor

This is the identity that Dataform uses to actually execute the BigQuery jobs defined in your SQLX files. This is a dedicated SA created for this solution.

Roles Required:

• BigQuery Data Editor: To create/modify tables and datasets.
• BigQuery Job User: To run BigQuery jobs.

3. Dataform Service Agent

The standard Dataform service agent needs permission to impersonate your executor service account. It requires the Service Account Token Creator role on your Dataform Workflow Executor SA.

The Cloud Function Code

The core logic resides in a Python Cloud Function. This function needs to:

1. Authenticate.
2. Find the correct compilation result for your environment (e.g., “production” or “test”).
3. Trigger the workflow invocation.

Here is a clean implementation using the google-cloud-dataform v1 library.

requirements.txt

google-cloud-dataform>=0.7.0
functions-framework>=3.0.0

main.py

import os
import google.auth
from google.cloud import dataform_v1

def trigger_dataform_workflow(cloudevent):
    """
    Triggered by a CloudEvent from Eventarc on GCS file changes.
    Calls the Dataform API to start a workflow invocation.
    """
    print(f"Received CloudEvent: {cloudevent.data}")

    try:
        # 1. Load Configuration
        PROJECT_ID = os.environ.get("GCP_PROJECT")
        PROJECT_NUMBER = os.environ.get("GCP_PROJECT_NUMBER")
        LOCATION = os.environ.get("DATAFORM_LOCATION")
        REPOSITORY_ID = os.environ.get("DATAFORM_REPOSITORY_ID")
        DATAFORM_RUNNER_SA = os.environ.get("DATAFORM_RUNNER_SA")
        DATAFORM_RELEASE_CONFIG_ID = os.environ.get("DATAFORM_RELEASE_CONFIG_ID")

        if not all([PROJECT_ID, PROJECT_NUMBER, LOCATION, REPOSITORY_ID, DATAFORM_RUNNER_SA, DATAFORM_RELEASE_CONFIG_ID]):
            print("Error: Missing one or more required environment variables")
            return "Error: Missing environment variables", 500

        # 2. Authenticate and Initialize Client
        credentials, _ = google.auth.default()
        dataform_client = dataform_v1.DataformClient(credentials=credentials)

        parent = dataform_client.repository_path(PROJECT_ID, LOCATION, REPOSITORY_ID)

        # 3. Find the Latest Compilation Result
        # We construct the full resource name to match the release config
        release_config_name_to_match = f"projects/{PROJECT_NUMBER}/locations/{LOCATION}/repositories/{REPOSITORY_ID}/releaseConfigs/{DATAFORM_RELEASE_CONFIG_ID}"
        print(f"Looking for Compilation Results matching: {release_config_name_to_match}")

        compilation_results = dataform_client.list_compilation_results(parent=parent)

        latest_compilation_name = None
        latest_timestamp = None

        for result in compilation_results:
            if result.release_config == release_config_name_to_match:
                current_create_time = result.create_time
                if latest_timestamp is None or current_create_time > latest_timestamp:
                    latest_timestamp = current_create_time
                    latest_compilation_name = result.name

        if not latest_compilation_name:
            print(f"Error: No successful compilation result found for {release_config_name_to_match}")
            return f"Error: No usable compilation result found", 404

        print(f"Using Compilation Result: {latest_compilation_name}")

        # 4. Create Workflow Invocation
        workflow_invocation = dataform_v1.WorkflowInvocation()
        workflow_invocation.compilation_result = latest_compilation_name
        workflow_invocation.invocation_config = dataform_v1.InvocationConfig(
            service_account=DATAFORM_RUNNER_SA
        )

        request = dataform_v1.CreateWorkflowInvocationRequest(
            parent=parent,
            workflow_invocation=workflow_invocation,
        )

        print(f"Targeting Parent Resource: {parent}")
        print(f"Running as Service Account: {DATAFORM_RUNNER_SA}")

        response = dataform_client.create_workflow_invocation(request=request)
        print(f"Successfully created Dataform workflow invocation: {response.name}")

        return "Dataform workflow invocation created", 200

    except Exception as e:
        print(f"Error triggering Dataform workflow: {e}")
        return f"Error: {e}", 500

Environment Variables

To make this code reusable across environments, ensure you set the following environment variables when deploying your Cloud Function:

• GCP_PROJECT: Your Google Cloud Project ID.
• GCP_PROJECT_NUMBER: Your Project Number.
• DATAFORM_LOCATION: The region where your Dataform repository is located.
• DATAFORM_REPOSITORY_ID: The name of your repository.
• DATAFORM_RUNNER_SA: The email of the service account that will execute the SQL.
• DATAFORM_RELEASE_CONFIG_ID: The specific release configuration (e.g., production or staging) you want to target.

Conclusion

By coupling Eventarc with Cloud Functions, you can create a robust, serverless trigger for your Dataform pipelines. This ensures your data transformations kick off the moment your raw data arrives, keeping your analytics fresh and your latency low.

Happy coding!