Interroger pour savoir si le job BigQuery est terminé

Lorsque vous lancez une tâche BigQuery à l'aide du connecteur BigQuery, il est possible que votre workflow n'attende pas automatiquement la fin de la tâche. Si les étapes suivantes dépendent de la réussite du job, vous devez implémenter un mécanisme d'interrogation.

De nombreux connecteurs Workflows renvoient une ressource d'opération de longue durée (LRO) qui peut être interrogée. Alors que certains connecteurs renvoient une ressource de type Operation pour représenter une LRO, BigQuery utilise une ressource Job à cet effet. D'autres services, tels que Cloud Build ou Compute Engine, renvoient une ressource Operation réelle. Pour obtenir la liste des types de ressources LRO, consultez Connecteurs compatibles.

Détection automatique

Vous pouvez interroger automatiquement l'état d'une tâche BigQuery lorsque vous utilisez des méthodes de connecteur qui renvoient une ressource LRO, comme googleapis.bigquery.v2.jobs.insert.

Lorsque vous utilisez une telle méthode de connecteur, Workflows bloque l'exécution du workflow jusqu'à ce que l'opération aboutisse ou échoue. Le champ connector_params vous permet de personnaliser le comportement d'interrogation, y compris le délai d'attente total et la fréquence des vérifications d'état. Pour en savoir plus, consultez Appeler un connecteur.

Par exemple, le workflow suivant montre comment démarrer une tâche BigQuery à l'aide de la méthode jobs.insert et comment configurer l'interrogation de la tâche avec une stratégie qui utilise des paramètres personnalisés plutôt que les paramètres par défaut :

YAML

# This workflow demonstrates how to automatically poll for the completion of
# a BigQuery job when using a connector method that return an LRO resource.
main:
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - query: "SELECT * FROM `bigquery-public-data.usa_names.usa_1910_2013` LIMIT 10"
    - run_bigquery_job:
        call: googleapis.bigquery.v2.jobs.insert
        args:
          projectId: ${project_id}
          body:
            configuration:
              query:
                query: ${query}
                useLegacySql: false
          connector_params:
            timeout: 3600  # Total time in seconds to wait for the job; default is 1800
            polling_policy:
              initial_delay: 2.0  # Seconds to wait before the first poll; default is 1.0
              multiplier: 1.5     # Factor by which to increase delay between polls; default is 1.25
              max_delay: 60.0     # Maximum delay in seconds between polls; default is 60.0
        result: job_status
    - finish:
        return: ${job_status}

JSON

{
  "main": {
    "steps": [
      {
        "init": {
          "assign": [
            {
              "project_id": "${sys.get_env(\"GOOGLE_CLOUD_PROJECT_ID\")}"
            },
            {
              "query": "SELECT * FROM `bigquery-public-data.usa_names.usa_1910_2013` LIMIT 10"
            }
          ]
        }
      },
      {
        "run_bigquery_job": {
          "call": "googleapis.bigquery.v2.jobs.insert",
          "args": {
            "projectId": "${project_id}",
            "body": {
              "configuration": {
                "query": {
                  "query": "${query}",
                  "useLegacySql": false
                }
              }
            },
            "connector_params": {
              "timeout": 3600,
              "polling_policy": {
                "initial_delay": 2,
                "multiplier": 1.5,
                "max_delay": 60
              }
            }
          },
          "result": "job_status"
        }
      },
      {
        "finish": {
          "return": "${job_status}"
        }
      }
    ]
  }
}

Détection manuelle

L'interrogation automatique ne s'applique qu'aux méthodes qui renvoient une ressource LRO. Les méthodes synchrones standards telles que googleapis.bigquery.v2.jobs.query ne sont pas compatibles avec l'interrogation automatique et peuvent nécessiter une interrogation manuelle si la requête prend plus de temps que le délai d'expiration interne de l'API.

Par exemple, le workflow suivant montre comment interroger l'état d'une tâche BigQuery directement dans le workflow à l'aide de la méthode getQueryResults :

YAML

# This workflow demonstrates how to manually poll for the completion of
# a BigQuery job by using the `getQueryResults` method.
main:
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
    - startQuery:
        call: googleapis.bigquery.v2.jobs.query
        args:
          projectId: ${project_id}
          body:
            useLegacySql: false
            query: "SELECT * FROM `bigquery-public-data.usa_names.usa_1910_2013` LIMIT 10"
        result: queryResponse
    - getJobId:
        assign:
          - jobId: ${queryResponse.jobReference.jobId}
    - pollingLoop:
        steps:
          - checkStatus:
              call: googleapis.bigquery.v2.jobs.getQueryResults
              args:
                projectId: ${project_id}
                jobId: ${jobId}
                timeoutMs: 10000  # Default wait time per call
              result: jobStatus
          - checkIfDone:
              switch:
                - condition: ${jobStatus.jobComplete}
                  return: ${jobStatus}  # Job is finished
          - wait:
              call: sys.sleep
              args:
                seconds: 5
              next: checkStatus  # Repeat check

JSON

{
  "main": {
    "steps": [
      {
        "init": {
          "assign": [
            {
              "project_id": "${sys.get_env(\"GOOGLE_CLOUD_PROJECT_ID\")}"
            }
          ]
        }
      },
      {
        "startQuery": {
          "call": "googleapis.bigquery.v2.jobs.query",
          "args": {
            "projectId": "${project_id}",
            "body": {
              "useLegacySql": false,
              "query": "SELECT * FROM `bigquery-public-data.usa_names.usa_1910_2013` LIMIT 10"
            }
          },
          "result": "queryResponse"
        }
      },
      {
        "getJobId": {
          "assign": [
            {
              "jobId": "${queryResponse.jobReference.jobId}"
            }
          ]
        }
      },
      {
        "pollingLoop": {
          "steps": [
            {
              "checkStatus": {
                "call": "googleapis.bigquery.v2.jobs.getQueryResults",
                "args": {
                  "projectId": "${project_id}",
                  "jobId": "${jobId}",
                  "timeoutMs": 10000
                },
                "result": "jobStatus"
              }
            },
            {
              "checkIfDone": {
                "switch": [
                  {
                    "condition": "${jobStatus.jobComplete}",
                    "return": "${jobStatus}"
                  }
                ]
              }
            },
            {
              "wait": {
                "call": "sys.sleep",
                "args": {
                  "seconds": 5
                },
                "next": "checkStatus"
              }
            }
          ]
        }
      }
    ]
  }
}

Étapes suivantes