BigQuery 작업 완료 여부 폴링

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

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

자동 폴링

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

이러한 커넥터 메서드를 사용하면 작업이 성공하거나 실패할 때까지 워크플로 실행이 차단됩니다. 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과 같은 표준 동기 메서드는 자동 폴링을 지원하지 않으며 쿼리가 내부 API 제한 시간보다 오래 걸리는 경우 수동 폴링이 필요할 수 있습니다.

예를 들어 다음 워크플로는 getQueryResults 메서드를 사용하여 워크플로 내에서 직접 BigQuery 작업의 완료를 폴링하는 방법을 보여줍니다.

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

다음 단계