Solved: Dataform workflow orchestration - Page 2

frans-snyders · 10-11-2022 05:46 AM

I have set up workflow to run a specific tag in a branch as per the guide here, but I'm seeing that the workflow triggers a new reload regardless of the state of the previous reload.

I was expecting Workflow or Dataform to wait for the existing task to complete.

Is there a way to configure the Workflow or Dataform to

skip a reload if the same reload already exists
retry a reload ?

Maybe by using an API call in workflow to get the state of the last execution ?

ghh

Thanks for the clarification. Dataform will indeed re-create/update all tables every time a Workflow is executed.

In Cloud Workflows, you can add a condition that checks if there are invocations running with a particular tag and end the workflow if that's the case. We'll also add that option when we release a scheduler in the Dataform app.

Here's an example:

main:
    steps:
    - init:
        assign:
          - repository: projects/PROJECT_ID/locations/REPOSITORY_LOCATION/repositories/REPOSITORY_ID
    - current_invocations
      call: http.get
      args:
        url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/workflowInvocations?filter=invocationConfig.includedTags:daily&filter=state:RUNNING"}
        auth:
          type: OAuth2
    result: running_invocations
    - decide_to_run:
      switch:
        - condition: ${len(running_invocations.body) > 0}
          next: end
    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
        result: compilationResult
    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
        result: workflowInvocation
    - complete:
        return: ${workflowInvocation.body.name}

EDIT: updated code snippet with reply below

View solution in original post