BigQuery 작업 완료 여부 폴링

BigQuery 커넥터를 사용하여 BigQuery 작업을 시작할 때, 워크플로가 작업 완료를 자동으로 기다리지 않을 수 있습니다. 후속 단계가 작업의 성공 여부에 따라 달라지는 경우 폴링 메커니즘을 구현해야 합니다.

많은 Workflows 커넥터는 폴링할 수 있는 장기 실행 작업 (LRO) 리소스를 반환합니다. 일부 커넥터는 LRO를 나타내는 Operation 유형의 리소스를 반환하지만 BigQuery는 이 용도로 Job 리소스를 사용합니다. Cloud Build 또는 Compute Engine과 같은 다른 서비스는 실제 Operation 리소스를 반환합니다. LRO 리소스 유형 목록은 지원되는 커넥터를 참조하세요.

자동 폴링

`googleapis.bigquery.v2.jobs.insert`와 같이 LRO 리소스를 반환하는 커넥터 메서드를 사용할 때 BigQuery 작업의 완료를 자동으로 폴링할 수 있습니다.

이러한 커넥터 메서드를 사용하면 작업이 성공하거나 실패할 때까지 Workflows가 워크플로 실행을 차단합니다. connector_params 필드를 사용하면 총 제한 시간 및 상태 확인 빈도를 비롯한 폴링 동작을 맞춤설정할 수 있습니다. 자세한 내용은 커넥터 호출을 참조하세요.

예를 들어 다음 워크플로는 jobs.insert 메서드를 사용하여 BigQuery 작업을 시작하고 기본 설정이 아닌 커스텀 설정을 사용하는 정책으로 작업의 폴링을 구성하는 방법을 보여줍니다.

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}"
        }
      }
    ]
  }
}

수동 폴링

자동 폴링은 LRO 리소스를 반환하는 메서드에만 적용됩니다. `googleapis.bigquery.v2.jobs.query`와 같은 표준 동기 메서드는 googleapis.bigquery.v2.jobs.query 자동 폴링을 지원하지 않으며 쿼리가 내부 API 제한 시간보다 오래 걸리는 경우 수동 폴링이 필요할 수 있습니다.

예를 들어 다음 워크플로는 워크플로 내에서 직접 BigQuery 작업의 완료를 폴링하는 방법을 보여줍니다. 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"
              }
            }
          ]
        }
      }
    ]
  }
}

다음 단계