연속 쿼리 모니터링

다음 BigQuery 도구를 사용하여 BigQuery 연속 쿼리를 모니터링할 수 있습니다.

• INFORMATION_SCHEMA 뷰
• 쿼리 실행 그래프
• 작업 기록
• 관리 작업 탐색기

BigQuery 연속 쿼리의 장기 실행 특성으로 인해 일반적으로 SQL 쿼리가 완료되면 생성되는 측정항목이 없거나 부정확할 수 있습니다.

INFORMATION_SCHEMA 보기 사용

여러 INFORMATION_SCHEMA 뷰를 사용하여 연속 쿼리와 연속 쿼리 예약을 모니터링할 수 있습니다.

참고: 실행 중인 연속 쿼리의 데이터는 JOBS* INFORMATION_SCHEMA 뷰에서 2일 동안만 보관됩니다.

작업 세부정보 보기

JOBS 뷰를 사용하여 연속 쿼리 작업 메타데이터를 가져올 수 있습니다.

다음 쿼리는 활성 상태인 모든 연속 쿼리의 메타데이터를 반환합니다. 메타데이터에는 연속 쿼리가 데이터를 성공적으로 처리한 시점을 나타내는 출력 워터마크 타임스탬프가 포함됩니다.

1. Google Cloud 콘솔에서 BigQuery 페이지로 이동합니다.

   BigQuery로 이동

2. 쿼리 편집기 창에서 다음 쿼리를 실행합니다.

   SELECT
     start_time,
     job_id,
     user_email,
     query,
     state,
     reservation_id,
     continuous_query_info.output_watermark
   FROM `PROJECT_ID.region-REGION.INFORMATION_SCHEMA.JOBS`
   WHERE
     creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 day)
     AND continuous IS TRUE
     AND state = "RUNNING"
   ORDER BY
     start_time DESC

   다음을 바꿉니다.

   • PROJECT_ID: 프로젝트 ID
   • REGION: 모든 데이터 세트 리전 이름입니다. 예를 들면 region-us입니다.

예약 할당 세부정보 보기

ASSIGNMENTS 및 RESERVATIONS 뷰를 사용하여 연속 쿼리 예약 할당 세부정보를 가져올 수 있습니다.

연속 쿼리의 예약 할당 세부정보를 반환합니다.

1. Google Cloud 콘솔에서 BigQuery 페이지로 이동합니다.

   BigQuery로 이동

2. 쿼리 편집기 창에서 다음 쿼리를 실행합니다.

   SELECT
     reservation.reservation_name,
     reservation.slot_capacity
   FROM
     `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.ASSIGNMENTS`
       AS assignment
   INNER JOIN
     `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.RESERVATIONS`
       AS reservation
     ON (assignment.reservation_name = reservation.reservation_name)
   WHERE
     assignment.assignee_id = 'PROJECT_ID'
     AND job_type = 'CONTINUOUS';

   다음을 바꿉니다.

   • ADMIN_PROJECT_ID: 예약을 소유한 관리 프로젝트의 ID
   • LOCATION: 예약 위치
   • PROJECT_ID: 예약에 할당된 프로젝트의 ID. 이 프로젝트에서 실행 중인 연속 쿼리에 관한 정보만 반환됩니다.

슬롯 사용량 정보 보기

ASSIGNMENTS, RESERVATIONS, JOBS_TIMELINE 뷰를 사용하여 연속 쿼리 슬롯 사용량 정보를 가져올 수 있습니다.

연속 쿼리의 슬롯 사용량 정보를 반환합니다.

1. Google Cloud 콘솔에서 BigQuery 페이지로 이동합니다.

   BigQuery로 이동

2. 쿼리 편집기 창에서 다음 쿼리를 실행합니다.

   SELECT
     jobs.period_start,
     reservation.reservation_name,
     reservation.slot_capacity,
     SUM(jobs.period_slot_ms) / 1000 AS consumed_total_slots
   FROM
     `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.ASSIGNMENTS`
       AS assignment
   INNER JOIN
     `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.RESERVATIONS`
       AS reservation
     ON (assignment.reservation_name = reservation.reservation_name)
   INNER JOIN
     `PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.JOBS_TIMELINE` AS jobs
     ON (
       UPPER(CONCAT('ADMIN_PROJECT_ID:LOCATION.', assignment.reservation_name))
       = UPPER(jobs.reservation_id))
   WHERE
     assignment.assignee_id = 'PROJECT_ID'
     AND assignment.job_type = 'CONTINUOUS'
     AND jobs.period_start
       BETWEEN TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
       AND CURRENT_TIMESTAMP()
   GROUP BY 1, 2, 3
   ORDER BY jobs.period_start DESC;

   다음을 바꿉니다.

   • ADMIN_PROJECT_ID: 예약을 소유한 관리 프로젝트의 ID
   • LOCATION: 예약 위치
   • PROJECT_ID: 예약에 할당된 프로젝트의 ID. 이 프로젝트에서 실행 중인 연속 쿼리에 관한 정보만 반환됩니다.

측정항목 탐색기 및 관리 리소스 차트와 같은 다른 도구를 사용하여 연속 쿼리 예약을 모니터링할 수도 있습니다. 자세한 내용은 BigQuery 예약 모니터링을 참고하세요.

쿼리 실행 그래프 사용

쿼리 실행 그래프를 사용하여 연속 쿼리에 대한 성능 통계와 일반 통계를 얻을 수 있습니다. 자세한 내용은 쿼리 성능 통계 보기를 참고하세요.

작업 기록 보기

개인 작업 기록 또는 프로젝트의 작업 기록에서 연속 쿼리 작업 세부정보를 볼 수 있습니다. 자세한 내용은 작업 세부정보 보기를 참조하세요.

이전 작업 목록은 작업 시작 시간을 기준으로 정렬되므로 한동안 실행된 연속 쿼리가 목록의 시작 시간과 근접하지 않을 수 있습니다.

관리 작업 탐색기 사용

관리 작업 탐색기에서 작업 카테고리 필터를 연속 쿼리로 설정하여 작업을 필터링하여 연속 쿼리를 표시합니다.

Cloud Monitoring 사용

Cloud Monitoring을 사용하여 BigQuery 연속 쿼리 관련 측정항목을 볼 수 있습니다. 자세한 내용은 대시보드, 차트, 알림 만들기를 참고하고 시각화에 사용할 수 있는 측정항목에 대해 알아보세요.

실패한 쿼리에 대한 알림

연속 쿼리가 실패했는지 정기적으로 확인하는 대신 실패를 알리는 알림을 만드는 것이 유용할 수 있습니다. 이를 수행하는 한 가지 방법은 작업 필터가 있는 맞춤 Cloud Logging 로그 기반 측정항목과 해당 측정항목을 기반으로 하는 Cloud Monitoring 알림 정책을 만드는 것입니다.

1. 연속 쿼리를 만들 때는 맞춤 작업 ID 접두사를 사용하세요. 여러 연속 쿼리가 동일한 접두사를 공유할 수 있습니다. 예를 들어 prod- 접두사를 사용하여 프로덕션 쿼리를 나타낼 수 있습니다.

2. Google Cloud 콘솔에서 로그 기반 측정항목 페이지로 이동합니다.

   로그 기반 측정항목으로 이동

3. 측정항목 만들기를 클릭합니다. 로그 측정항목 만들기 패널이 나타납니다.

4. 측정항목 유형에서 카운터를 선택합니다.

5. 세부정보 섹션에서 측정항목 이름을 지정합니다. 예를 들면 CUSTOM_JOB_ID_PREFIX-metric입니다.

6. 필터 선택 섹션의 필터 빌드 편집기에 다음을 입력합니다.

   resource.type = "bigquery_project"
   protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
   severity = ERROR
   

   다음을 바꿉니다.

   • PROJECT_ID: 프로젝트의 이름
   • CUSTOM_JOB_ID_PREFIX: 연속 쿼리에 설정한 맞춤 작업 ID 접두사의 이름입니다.

7. 측정항목 만들기를 클릭합니다.

8. 탐색 메뉴에서 로그 기반 측정항목을 클릭합니다. 방금 만든 측정항목이 사용자 정의 측정항목 목록에 표시됩니다.

9. 측정항목 행에서 작업 더보기를 클릭한 다음 측정항목에서 알림 만들기를 클릭합니다.

10. 다음을 클릭합니다. 정책 구성 모드 페이지에서 기본 설정을 변경할 필요가 없습니다.

11. 다음을 클릭합니다. 알림 트리거 구성 페이지에서 기본 설정을 변경할 필요는 없습니다.

12. 알림 채널을 선택하고 알림 정책의 이름을 입력합니다.

13. 정책 만들기를 클릭합니다.

선택한 맞춤 작업 ID 접두사로 연속 쿼리를 실행한 후 취소하여 알림을 테스트할 수 있습니다. 알림이 알림 채널에 도달하는 데 몇 분 정도 걸릴 수 있습니다.

실패한 쿼리 재시도

실패한 연속 쿼리를 다시 시도하면 연속 파이프라인이 장시간 다운되거나 다시 시작하기 위해 사람의 개입이 필요한 상황을 방지할 수 있습니다. 실패한 연속 쿼리를 다시 시도할 때 고려해야 할 중요한 사항은 다음과 같습니다.

• 이전 쿼리가 실패하기 전에 처리된 일부 데이터를 다시 처리해도 되는지 여부입니다.
• 재시도를 제한하거나 지수 백오프를 사용하는 방법

쿼리 재시도를 자동화하는 한 가지 방법은 다음과 같습니다.

1. 다음 기준을 충족하는 포함 필터를 기반으로 Cloud Logging 싱크를 만들어 로그를 Pub/Sub 주제로 라우팅합니다.

   resource.type = "bigquery_project"
   protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
   severity = ERROR
   

   다음을 바꿉니다.

   • PROJECT_ID: 프로젝트의 이름
   • CUSTOM_JOB_ID_PREFIX: 연속 쿼리에 설정한 맞춤 작업 ID 접두사의 이름입니다.

2. 필터와 일치하는 로그를 수신하는 Pub/Sub에 대한 응답으로 트리거되는 Cloud Run 함수를 만듭니다.

   Cloud Run 함수는 Pub/Sub 메시지에서 데이터 페이로드를 수락하고 실패한 쿼리와 동일한 SQL 구문을 사용하여 이전 작업이 중지된 직후에 새 연속 쿼리를 시작하려고 시도할 수 있습니다.

예를 들어 다음과 유사한 함수를 사용할 수 있습니다.

Python

이 샘플을 사용해 보기 전에 BigQuery 빠른 시작: 클라이언트 라이브러리 사용의 Python 설정 안내를 따르세요. 자세한 내용은 BigQuery Python API 참고 문서를 확인하세요.

BigQuery에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 클라이언트 라이브러리의 인증 설정을 참조하세요.

import base64
import json
import logging
import re
import uuid

import google.auth
import google.auth.transport.requests
import requests


def retry_continuous_query(event, context):
    logging.info("Cloud Function started.")

    if "data" not in event:
        logging.info("No data in Pub/Sub message.")
        return

    try:
        # Decode and parse the Pub/Sub message data
        log_entry = json.loads(base64.b64decode(event["data"]).decode("utf-8"))

        # Extract the SQL query and other necessary data
        proto_payload = log_entry.get("protoPayload", {})
        metadata = proto_payload.get("metadata", {})
        job_change = metadata.get("jobChange", {})
        job = job_change.get("job", {})
        job_config = job.get("jobConfig", {})
        query_config = job_config.get("queryConfig", {})
        sql_query = query_config.get("query")
        job_stats = job.get("jobStats", {})
        end_timestamp = job_stats.get("endTime")
        failed_job_id = job.get("jobName")

        # Check if required fields are missing
        if not all([sql_query, failed_job_id, end_timestamp]):
            logging.error("Required fields missing from log entry.")
            return

        logging.info(f"Retrying failed job: {failed_job_id}")

        # Adjust the timestamp in the SQL query
        timestamp_match = re.search(
            r"\s*TIMESTAMP\(('.*?')\)(\s*\+ INTERVAL 1 MICROSECOND)?", sql_query
        )

        if timestamp_match:
            original_timestamp = timestamp_match.group(1)
            new_timestamp = f"'{end_timestamp}'"
            sql_query = sql_query.replace(original_timestamp, new_timestamp)
        elif "CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE" in sql_query:
            new_timestamp = f"TIMESTAMP('{end_timestamp}') + INTERVAL 1 MICROSECOND"
            sql_query = sql_query.replace(
                "CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE", new_timestamp
            )

        # Get access token
        credentials, project = google.auth.default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)
        access_token = credentials.token

        # API endpoint
        url = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project}/jobs"

        # Request headers
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
        }

        # Generate a random UUID
        random_suffix = str(uuid.uuid4())[:8]  # Take the first 8 characters of the UUID

        # Combine the prefix and random suffix
        job_id = f"CUSTOM_JOB_ID_PREFIX{random_suffix}"

        # Request payload
        data = {
            "configuration": {
                "query": {
                    "query": sql_query,
                    "useLegacySql": False,
                    "continuous": True,
                    "connectionProperties": [
                        {"key": "service_account", "value": "SERVICE_ACCOUNT"}
                    ],
                    # ... other query parameters ...
                },
                "labels": {"bqux_job_id_prefix": "CUSTOM_JOB_ID_PREFIX"},
            },
            "jobReference": {
                "projectId": project,
                "jobId": job_id,  # Use the generated job ID here
            },
        }

        # Make the API request
        response = requests.post(url, headers=headers, json=data)

        # Handle the response
        if response.status_code == 200:
            logging.info("Query job successfully created.")
        else:
            logging.error(f"Error creating query job: {response.text}")

    except Exception as e:
        logging.error(
            f"Error processing log entry or retrying query: {e}", exc_info=True
        )

    logging.info("Cloud Function finished.")

다음 단계

• 연속 쿼리를 만들고 실행하는 방법을 알아봅니다.