監控持續查詢

您可以使用下列 BigQuery 工具，監控 BigQuery 持續查詢：

• INFORMATION_SCHEMA 次觀看
• 查詢執行圖表
• 工作記錄
• 工作探索工具

由於 BigQuery 持續查詢會長時間執行，因此通常在 SQL 查詢完成時產生的指標可能會遺失或不準確。

使用 INFORMATION_SCHEMA 檢視畫面

您可以使用多個 INFORMATION_SCHEMA 檢視表，監控持續查詢和持續查詢預留項目。

查看工作詳細資料

您可以使用 JOBS 檢視表，取得持續查詢工作的中繼資料。

下列查詢會傳回所有執行中持續查詢的中繼資料。中繼資料包含輸出浮水印時間戳記，代表持續查詢成功處理資料的時間點。

1. 前往 Cloud de Confiance 控制台的「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. 前往 Cloud de Confiance 控制台的「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. 前往 Cloud de Confiance 控制台的「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。系統只會傳回這項專案中執行的持續查詢相關資訊。

您也可以使用 Metrics Explorer 和管理資源圖表等其他工具，監控持續查詢預留項目。詳情請參閱「監控 BigQuery 預留項目」。

使用查詢執行圖

您可以使用查詢執行圖，取得持續查詢的效能深入分析和一般統計資料。詳情請參閱「查看查詢效能深入分析」。

查看工作記錄

您可以在個人工作記錄或專案的工作記錄中，查看持續查詢工作詳細資料。詳情請參閱「查看工作詳細資料」。

請注意，工作歷史記錄清單會依工作開始時間排序，因此已執行一段時間的持續查詢可能不會排在清單開頭。

使用工作探索工具

在工作探索工具中，將「Job category」(工作類別) 篩選器設為「Continuous query」(持續查詢)，篩選工作來顯示持續查詢。

使用 Cloud Monitoring

您可以使用 Cloud Monitoring 查看 BigQuery 持續查詢的特定指標。詳情請參閱「建立資訊主頁、圖表和快訊」，並瞭解可視化的指標。

查詢失敗時發出警告

與其定期檢查持續查詢是否失敗，不如建立快訊，在失敗時通知您。其中一種方法是建立自訂 Cloud Logging 記錄指標，並為工作設定篩選器，然後根據該指標建立 Cloud Monitoring 警告政策：

1. 建立持續查詢時，請使用自訂工作 ID 前置字串。多個持續查詢可以共用相同的前置字串。例如，您可以使用前置字串 prod- 來表示正式版查詢。

2. 前往 Cloud de Confiance 控制台的「Log-based Metrics」(記錄指標) 頁面。

   前往「Log-based Metrics」(記錄指標)

3. 點選「建立指標」。「建立記錄指標」面板隨即顯示。

4. 在「Metric type」(指標類型) 部分，選取「Counter」(計數器)。

5. 在「詳細資料」部分，為指標命名。例如：CUSTOM_JOB_ID_PREFIX-metric。

6. 在「Filter selection」(選取篩選條件) 區段中，在「Build filter」(建立篩選器) 編輯器中輸入下列內容：

   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. 在導覽選單中，按一下「Log-based metrics」(記錄指標)。您剛建立的指標會顯示在使用者定義指標清單中。

9. 在指標資料列中，依序點選 more_vert「More actions」(更多動作) 和「Create alert from metric」(運用指標建立警告)。

10. 點選「下一步」。您不需要變更「Policy configuration mode」(政策設定模式) 頁面上的預設設定。

11. 點選「下一步」。您不需要變更「Configure alert trigger」(設定警告觸發條件) 頁面上的預設設定。

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. 建立 Cloud Run 函式，在 Pub/Sub 收到符合篩選條件的記錄時觸發。

   Cloud Run 函式可接受 Pub/Sub 訊息中的資料酬載，並嘗試使用與失敗查詢相同的 SQL 語法，啟動新的持續查詢，但從前一個工作停止後的時間點開始。

例如，您可以使用類似下列的函式：

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.")

後續步驟

• 瞭解如何建立及執行持續查詢。