BigQuery Scheduled Query Observability Stack —

BigQuery Observability

BigQuery Scheduled Query Observability Stack Monitoring Alerting Logging Dashboard Data Pipeline ETL Production

Scheduled Query Setup

# === BigQuery Scheduled Query with Terraform ===



# resource "google_bigquery_data_transfer_config" "daily_etl" {

#   display_name           = "daily_orders_etl"

#   data_source_id         = "scheduled_query"

#   schedule               = "every day 02:00"

#   destination_dataset_id = "analytics"

#   location               = "asia-southeast1"

#

#   params = {

#     destination_table_name_template = "orders_daily_{run_date}"

#     write_disposition               = "WRITE_TRUNCATE"

#     query                           = <<-SQL

#       SELECT

#         DATE(created_at) as order_date,

#         COUNT(*) as order_count,

#         SUM(total_amount) as total_revenue,

#         COUNT(DISTINCT customer_id) as unique_customers

#       FROM `project.raw.orders`

#       WHERE DATE(created_at) = @run_date

#       GROUP BY 1

#     SQL

#   }

#

#   email_preferences {

#     enable_failure_email = true

#   }

# }



# bq CLI

# bq mk --transfer_config \

#   --target_dataset=analytics \

#   --display_name='daily_orders_etl' \

#   --data_source=scheduled_query \

#   --schedule='every day 02:00' \

#   --params='{

#     "query": "SELECT ... FROM ... WHERE DATE(created_at) = @run_date",

#     "destination_table_name_template": "orders_daily_{run_date}",

#     "write_disposition": "WRITE_TRUNCATE"

#   }'



from dataclasses import dataclass



@dataclass

class ScheduledQuery:

    name: str

    schedule: str

    purpose: str

    destination: str

    alert: str



queries = [

    ScheduledQuery("daily_orders_etl",

        "every day 02:00",

        "รวม Orders รายวัน Revenue Customers",

        "analytics.orders_daily_{run_date}",

        "Fail → Slack #data-alerts"),

    ScheduledQuery("hourly_metrics_agg",

        "every 1 hours",

        "รวม Application Metrics ทุกชั่วโมง",

        "analytics.app_metrics_hourly",

        "Fail + Duration > 10min → PagerDuty"),

    ScheduledQuery("data_quality_check",

        "every day 03:00 (หลัง ETL)",

        "ตรวจ Row Count NULL Freshness",

        "monitoring.dq_results",

        "Quality Fail → Slack + Jira Ticket"),

    ScheduledQuery("weekly_report",

        "every monday 06:00",

        "สร้าง Weekly Summary Report",

        "reports.weekly_summary",

        "Fail → Email to Data Team"),

]



print("=== Scheduled Queries ===")

for q in queries:

    print(f"  [{q.name}] Schedule: {q.schedule}")

    print(f"    Purpose: {q.purpose}")

    print(f"    Destination: {q.destination}")

    print(f"    Alert: {q.alert}")

Monitoring & Alerting

# === Observability Setup ===



# INFORMATION_SCHEMA query for job monitoring

# SELECT

#   job_id,

#   creation_time,

#   end_time,

#   TIMESTAMP_DIFF(end_time, start_time, SECOND) as duration_sec,

#   total_bytes_processed,

#   total_slot_ms,

#   state,

#   error_result.reason as error_reason,

#   error_result.message as error_message

# FROM `region-asia-southeast1`.INFORMATION_SCHEMA.JOBS

# WHERE job_type = 'QUERY'

#   AND creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)

# ORDER BY creation_time DESC

# LIMIT 100;



# Data Quality Check Query

# SELECT

#   'orders' as table_name,

#   COUNT(*) as row_count,

#   COUNTIF(customer_id IS NULL) as null_customer_id,

#   COUNTIF(total_amount IS NULL) as null_total_amount,

#   MAX(created_at) as latest_record,

#   TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), MAX(created_at), HOUR) as freshness_hours

# FROM `project.raw.orders`

# WHERE DATE(created_at) = CURRENT_DATE();



@dataclass

class AlertRule:

    alert: str

    condition: str

    severity: str

    channel: str

    runbook: str



alerts = [

    AlertRule("Scheduled Query Failed",

        "transfer_config state = FAILED",

        "Critical",

        "Slack #data-alerts + PagerDuty",

        "ตรวจ Error Message → แก้ Query → Manual Rerun"),

    AlertRule("Query Duration > SLA",

        "duration_sec > 2x average",

        "Warning",

        "Slack #data-alerts",

        "ตรวจ Query Plan → Optimize Partitioning/Clustering"),

    AlertRule("Bytes Processed Spike",

        "total_bytes_processed > 3x average",

        "Warning",

        "Slack #data-costs",

        "ตรวจ Full Table Scan → เพิ่ม WHERE Clause"),

    AlertRule("Data Quality Failure",

        "row_count = 0 OR null_rate > 5% OR freshness > 4hr",

        "Critical",

        "Slack + Jira Ticket",

        "ตรวจ Source Data → แจ้ง Data Producer"),

    AlertRule("Cost Budget Exceeded",

        "daily_bytes > budget_bytes",

        "Warning",

        "Email to Data Lead",

        "Review Expensive Queries → Optimize or Limit"),

]



print("=== Alert Rules ===")

for a in alerts:

    print(f"  [{a.alert}] Severity: {a.severity}")

    print(f"    Condition: {a.condition}")

    print(f"    Channel: {a.channel}")

    print(f"    Runbook: {a.runbook}")

Dashboard

# === Observability Dashboard ===



@dataclass

class DashboardPanel:

    panel: str

    source: str

    visualization: str

    refresh: str



panels = [

    DashboardPanel("Scheduled Query Status",

        "INFORMATION_SCHEMA.JOBS + Transfer Config API",

        "Status Table: Green=Success Red=Failed Yellow=Running",

        "ทุก 5 นาที"),

    DashboardPanel("Query Duration Trend",

        "INFORMATION_SCHEMA.JOBS duration_sec",

        "Line Chart: duration per query over time + SLA line",

        "ทุก 15 นาที"),

    DashboardPanel("Bytes Processed Daily",

        "INFORMATION_SCHEMA.JOBS total_bytes_processed",

        "Bar Chart: daily bytes + cost estimate + budget line",

        "ทุกชั่วโมง"),

    DashboardPanel("Slot Utilization",

        "Cloud Monitoring bigquery.googleapis.com/slots",

        "Area Chart: slot usage over time + capacity line",

        "ทุก 5 นาที"),

    DashboardPanel("Data Quality Scores",

        "monitoring.dq_results table",

        "Scorecard: per table completeness freshness accuracy",

        "หลังทุก DQ Check"),

    DashboardPanel("Error Log",

        "Cloud Logging BigQuery Audit Log",

        "Log Table: recent errors with details and job_id",

        "Real-time"),

]



print("=== Dashboard Panels ===")

for p in panels:

    print(f"  [{p.panel}] Source: {p.source}")

    print(f"    Viz: {p.visualization}")

    print(f"    Refresh: {p.refresh}")

เคล็ดลับ

• Alert: ตั้ง Alert ทุก Scheduled Query Fail เป็นอันดับแรก
• DQ: ตั้ง Data Quality Check หลังทุก ETL Job ทุกครั้ง
• Cost: Monitor Bytes Processed ป้องกัน Cost Spike
• Terraform: จัดการ Scheduled Query เป็น Code ใน Git
• INFORMATION_SCHEMA: ใช้วิเคราะห์ Performance ดีกว่า Console

การนำไปใช้งานจริงในองค์กร

สำหรับองค์กรขนาดกลางถึงใหญ่ แนะนำให้ใช้หลัก Three-Tier Architecture คือ Core Layer ที่เป็นแกนกลางของระบบ Distribution Layer ที่ทำหน้าที่กระจาย Traffic และ Access Layer ที่เชื่อมต่อกับผู้ใช้โดยตรง การแบ่ง Layer ชัดเจนช่วยให้การ Troubleshoot ง่ายขึ้นและสามารถ Scale ระบบได้ตามความต้องการ

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: A/B Testing ML Troubleshooting แก้ปัญหา

เรื่อง Network Security ก็สำคัญไม่แพ้กัน ควรติดตั้ง Next-Generation Firewall ที่สามารถ Deep Packet Inspection ได้ ใช้ Network Segmentation แยก VLAN สำหรับแต่ละแผนก ติดตั้ง IDS/IPS เพื่อตรวจจับการโจมตี และทำ Regular Security Audit อย่างน้อยปีละ 2 ครั้ง

สรุป

BigQuery Scheduled Query Observability Monitoring Alerting Logging Dashboard INFORMATION_SCHEMA Data Quality ETL Cost Control Production Pipeline

เนื้อหาเกี่ยวข้อง — RAM Pc3l คืออะไร — ข้อมูลครบถ้วน 2026