To run a Dataform pipeline using a specific tag and ensure that the execution is recorded in Dataform's execution history, you'll need to use the Dataform API correctly. The dataform_v1beta library provides the functionality to create workflow invocations and run pipelines, but some details must be carefully managed.
Below are some steps you can take and example script to run a Dataform workflow using a specific tag:
-
Install the Required Libraries: Ensure you have the google-cloud-dataform package installed:
pip install google-cloud-dataform
-
Authenticate Your Application: Set up authentication using a service account key file.
-
Initialize the Dataform Client: Create a DataformClient to interact with the Dataform API.
-
Obtain the Compilation Result ID: You must compile the repository to get a compilation_result_id. This compilation result is necessary to create a workflow invocation.
-
Create and Execute a Workflow Invocation: Use the create_workflow_invocation method to create and start the workflow invocation using the specific tag.
import os
from google.cloud import dataform_v1beta1
from google.oauth2 import service_account
# Set up your proxy if needed
os.environ['https_proxy'] = 'http://10.105.160.5:8080'
os.environ['http_proxy'] = 'http://10.105.160.5:8080'
def run_dataform_pipeline(project_id, location, repository_id, tag):
# Path to your service account key file
credentials_file = "P:/COMUM/FALCAO/30. EBB/09. Analistas ODC+CS+MS/br-apps-atendimentodva-qa-6ef8f7a85d97.json"
credentials = service_account.Credentials.from_service_account_file(credentials_file)
# Initialize the Dataform client
client = dataform_v1beta1.DataformClient(credentials=credentials)
# Define the parent path
parent = f"projects/{project_id}/locations/{location}/repositories/{repository_id}"
# Compile the repository to obtain the compilation result ID
compile_request = dataform_v1beta1.CompileRepositoryRequest(name=parent)
compilation_result = client.compile_repository(request=compile_request)
compilation_result_id = compilation_result.name.split('/')[-1]
# Define the workflow invocation with the specific tag
workflow_invocation = dataform_v1beta1.WorkflowInvocation(
compilation_result=f"{parent}/compilationResults/{compilation_result_id}",
invocation_config=dataform_v1beta1.WorkflowInvocation.InvocationConfig(
included_tags=[tag]
)
)
# Create the workflow invocation request
request = dataform_v1beta1.CreateWorkflowInvocationRequest(
parent=parent,
workflow_invocation=workflow_invocation
)
try:
# Run the workflow invocation
operation = client.create_workflow_invocation(request=request)
response = operation.result() # Wait for the operation to complete
print(f"WorkflowInvocation created: {response.name}")
except Exception as e:
print(f"Error running Dataform pipeline: {e}")
raise
# Example usage
project_id = "br-apps-atendimentodva-qa"
location = "us-central1"
repository_id = "Jasper"
tag = "base_ura_chamadas"
run_dataform_pipeline(project_id, location, repository_id, tag)
Additional Notes
-
Authentication: Ensure your environment is properly authenticated and that the service account has the necessary permissions.
-
Permissions: Verify that the service account used in the script has access to the Dataform repository and permissions to create workflow invocations.
-
Compilation and Invocation: Make sure to compile your Dataform repository before creating a workflow invocation.
-
Proxy Configuration: Include proxy settings only if required by your network setup.
Ensure you replace the placeholders (project_id, location, repository_id, tag) with your actual values. The script should then compile the repository and create a workflow invocation that runs the Dataform pipeline using the specified tag.
Feel free to modify the script based on your setup or add more detailed logging or error handling as needed.
LinkedIn
Twitter
