継続的クエリをモニタリングする

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。このプロジェクトで実行されている継続的クエリに関する情報のみが返されます。

Metrics Explorer や管理リソースのグラフなどのツールで継続的クエリの予約をモニタリングすることもできます。詳細については、BigQuery の予約をモニタリングするをご覧ください。

クエリ実行グラフを使用する

クエリ実行グラフを使用すると、継続的クエリのパフォーマンス分析情報と一般的な統計情報を取得できます。詳細については、クエリのパフォーマンス分析情報を表示するをご覧ください。

ジョブの履歴を表示する

継続的クエリジョブの詳細は、ユーザーのジョブ履歴またはプロジェクトのジョブ履歴で確認できます。詳細については、ジョブの詳細を表示するをご覧ください。

ジョブの履歴リストはジョブの開始時間で並べ替えられています。このため、長時間実行されている継続的クエリはリストの先頭のほうに表示されないことがあります。

管理ジョブ エクスプローラを使用する

管理ジョブ エクスプローラで、[ジョブのカテゴリ] フィルタを [継続的クエリ] に設定してジョブをフィルタリングし、継続的クエリを表示します。

Cloud Monitoring を使用する

BigQuery の継続的クエリに固有の指標を表示するには、Cloud Monitoring を使用します。詳細については、ダッシュボード、グラフ、アラートを作成すると表示に使用できる指標をご覧ください。

失敗したクエリに関するアラートを作成する

継続的クエリが失敗したかどうかを定期的に確認する代わりに、失敗について通知するアラートを作成すると便利です。これを行う方法の一つとして、ジョブのフィルタを使用してカスタムの 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 functions の関数を作成します。

   Cloud Run functions の関数は、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.")

次のステップ

• 継続的クエリを作成して実行する方法を確認する。