Quan sát pipeline: metric, log và bốn tín hiệu vàng cho dữ liệu

Pipeline đơn hàng của GDev Mart chạy lúc 02:00 mỗi đêm. Sáng thứ Hai, dashboard hiển thị doanh thu của thứ Năm tuần trước, tức bốn ngày dữ liệu cũ. Không có lỗi nào được báo: job Airflow vẫn "xanh" mỗi đêm, vì task cuối cùng chỉ ghi ra một bảng rỗng khi nguồn không có dữ liệu mới, và task đó vẫn thành công. Đội dữ liệu chỉ biết khi người dùng hỏi.

Hãy dự đoán trước khi đọc tiếp: bạn sẽ theo dõi những số đo nào của pipeline để biết nó khỏe hay không, ngoài câu "job có chạy thành công không"? Hãy liệt kê ít nhất ba số đo và nói mỗi số đo bắt được loại sự cố nào.

Sau bài này, bạn có thể

• Giải thích vì sao "job thành công" không chứng minh dữ liệu mới, và chọn freshness, volume, duration hay failure rate cho từng loại sự cố.
• Viết đoạn Python đẩy số đo lên Pushgateway sao cho thời điểm thành công chỉ cập nhật khi có dữ liệu ghi ra.
• Viết biểu thức PromQL tính freshness và chọn ngưỡng theo chu kỳ chạy của pipeline.
• Ghi log có cấu trúc dạng JSON để lọc theo trường khi điều tra sự cố.

Cần biết trước: bài "Độ tin cậy của DAG: retry, timeout, sensor và cảnh báo" (chương Airflow) và bài "Sáu chiều chất lượng dữ liệu và cách đo chúng" (chương chất lượng dữ liệu); để chạy ví dụ cần Python với thư viện prometheus_client.

Mô hình tư duy 1: quan sát được là hỏi hệ thống "bạn đang làm gì" mà không phải mở mã

Observability (khả năng quan sát) là mức độ bạn hiểu trạng thái bên trong của hệ thống chỉ từ những gì nó phát ra bên ngoài. Có ba loại tín hiệu bổ sung nhau:

Với pipeline dữ liệu, metric là tín hiệu để phát hiện, log là tín hiệu để tìm hiểu nguyên nhân.

Giải đáp: bốn tín hiệu vàng của pipeline dữ liệu

Các kỹ sư vận hành có "bốn tín hiệu vàng" cho dịch vụ web (độ trễ, lưu lượng, lỗi, độ bão hòa). Với pipeline dữ liệu, bốn tín hiệu tương ứng là:

Với sự cố đầu bài, freshness là tín hiệu quyết định: "lần chạy có dữ liệu thật gần nhất cách bây giờ 4 ngày". Metric "job thành công" không đủ vì job có thể thành công mà không làm gì có ích.

Cách đọc hình: bốn số đo chia thành hai nhóm (giá trị nghiệp vụ gồm freshness và volume, sức khỏe kỹ thuật gồm duration và failure rate), cùng chảy vào một nơi tập trung để đặt quy tắc cảnh báo và vẽ dashboard. Điểm mấu chốt: "thành công" là một sự kiện kỹ thuật, còn freshness và volume mới phản ánh giá trị nghiệp vụ.

Mô hình tư duy 2: Prometheus và Grafana, kéo số đo về một nơi

Prometheus là hệ thống lưu chuỗi thời gian và cảnh báo; Grafana vẽ dashboard từ nhiều nguồn, trong đó có Prometheus. Prometheus hoạt động theo mô hình kéo (pull): nó định kỳ truy cập địa chỉ /metrics của từng dịch vụ để lấy số đo. Hai loại metric cơ bản:

• Counter: chỉ tăng (tổng số dòng đã xử lý, tổng số lỗi); dùng rate() hoặc increase() để lấy tốc độ.
• Gauge: giá trị lên xuống (số dòng của lần chạy gần nhất, thời điểm thành công gần nhất).

Vấn đề với job chạy theo lô: nó chạy vài phút rồi tắt, nên Prometheus không kịp "kéo". Giải pháp là Pushgateway: job đẩy số đo cuối lần chạy sang Pushgateway, và Prometheus kéo từ Pushgateway. Pushgateway giữ giá trị cho tới khi bị ghi đè hoặc xóa (mặc định chỉ trong bộ nhớ, nên mất hết khi Pushgateway khởi động lại trừ khi bật --persistence.file); vì vậy đừng dựa vào "metric biến mất khi job tắt", hãy dùng gauge thời điểm thành công và so với giờ hiện tại.

import time

from prometheus_client import CollectorRegistry, Gauge, pushadd_to_gateway

registry = CollectorRegistry()
rows_in = Gauge("pipeline_rows_in", "Rows read from the source", ["pipeline"], registry=registry)
rows_out = Gauge("pipeline_rows_out", "Rows written to silver", ["pipeline"], registry=registry)
rows_rejected = Gauge("pipeline_rows_rejected", "Rows quarantined", ["pipeline"], registry=registry)
duration = Gauge("pipeline_duration_seconds", "Run duration", ["pipeline"], registry=registry)
last_success = Gauge("pipeline_last_success_timestamp_seconds", "Unix time of the last good run", ["pipeline"], registry=registry)


def publish(n_in, n_out, n_rejected, started):
    rows_in.labels("orders").set(n_in)
    rows_out.labels("orders").set(n_out)
    rows_rejected.labels("orders").set(n_rejected)
    duration.labels("orders").set(time.time() - started)
    if n_out > 0:
        last_success.labels("orders").set(time.time())
    pushadd_to_gateway("localhost:9091", job="gdevmart_orders", registry=registry)

Chú ý dòng if n_out > 0: chỉ cập nhật thời điểm thành công khi có dữ liệu thật. Đây chính là chỗ chặn sự cố đầu bài: một lần chạy ghi bảng rỗng không làm freshness "tươi lên". Vì vậy ví dụ dùng pushadd_to_gateway (HTTP POST, chỉ thay các metric có trong lần đẩy). Nếu dùng push_to_gateway (HTTP PUT), mỗi lần đẩy thay cả nhóm: lần chạy rỗng không đặt last_success nên chuỗi này biến mất hẳn thay vì giữ giá trị cũ, và quy tắc cảnh báo không còn gì để so sánh.

Công thức freshness trong PromQL: time() - pipeline_last_success_timestamp_seconds{pipeline="orders"} cho số giây kể từ lần thành công có dữ liệu. Với pipeline chạy một lần mỗi đêm, giá trị này tăng đều từ 0 tới khoảng 24 giờ ngay cả khi mọi thứ bình thường, nên ngưỡng phải lớn hơn chu kỳ chạy cộng dung sai, ví dụ 28 giờ (24 giờ cộng 4 giờ). Ngưỡng 6 giờ chỉ hợp lý cho pipeline chạy mỗi giờ.

Cách đọc hình: job ngắn đẩy số đo vào Pushgateway lúc kết thúc; Prometheus kéo từ đó định kỳ, lưu thành chuỗi thời gian; Grafana vẽ và Alertmanager gửi cảnh báo. Mỗi thành phần một việc, nối tiếp.

Log có cấu trúc để tìm nguyên nhân

Khi cảnh báo nổ, bạn cần log đủ ngữ cảnh. Thay vì print("error"), ghi log có trường cố định để lọc và truy vấn được:

import json
import logging
import sys

logger = logging.getLogger("pipeline")
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)


def log(event, **fields):
    logger.info(json.dumps({"event": event, **fields}, ensure_ascii=False))


log("run_finished", pipeline="orders", run_date="2026-09-02", rows_in=1010000, rows_out=990000, rows_rejected=20000, duration_s=412)

Mỗi dòng là một JSON; công cụ log (Loki, Elasticsearch, CloudWatch) có thể lọc event = "run_finished" AND rows_rejected > 50000. Hãy nhớ quy tắc từ tuần 2: không ghi dữ liệu cá nhân vào log.

Những hiểu lầm thường gặp

• "Job thành công nghĩa là dữ liệu tốt." Job có thể thành công khi ghi bảng rỗng; hãy đo freshness và volume.
• "Giám sát chỉ cần khi có sự cố." Số đo lịch sử cho phép phát hiện xu hướng xấu dần (job chậm dần mỗi tuần) trước khi nó thành sự cố.
• "Đẩy metric mà không có nhãn là đủ." Thiếu nhãn pipeline bạn không phân biệt được pipeline nào; nhưng đừng đưa giá trị có hàng triệu khả năng (như order_id) làm nhãn vì làm phình số chuỗi thời gian.
• "Pushgateway lưu mãi nên an toàn." Nó giữ giá trị cũ ngay cả khi job đã chết; luôn tính tuổi của số đo qua thời điểm thành công.

Tự kiểm tra (có đáp án)

Câu 1. Vì sao freshness bắt được sự cố đầu bài còn "job thành công" thì không? Đáp án: vì job vẫn thành công khi ghi bảng rỗng; freshness đo thời gian kể từ lần chạy có dữ liệu thật nên tăng dần khi nguồn không gửi gì.

Câu 2. Vì sao job chạy theo lô cần Pushgateway? Đáp án: vì job tắt sau vài phút nên Prometheus (mô hình kéo định kỳ) không kịp kéo; job đẩy số đo cuối lần chạy sang Pushgateway để Prometheus kéo từ đó.

Câu 3. Vì sao chỉ cập nhật last_success khi n_out > 0? Đáp án: để lần chạy ghi bảng rỗng không làm freshness trông tươi, nhờ đó cảnh báo "dữ liệu cũ" vẫn kích hoạt khi nguồn ngừng gửi.

Nếu chỉ nhớ 3 điều

1. Ba tín hiệu: metric để phát hiện, log để tìm nguyên nhân, trace cho hệ thống phân tán.
2. Bốn tín hiệu vàng của pipeline dữ liệu: freshness, volume, duration, failure rate; "job thành công" không đủ.
3. Job theo lô đẩy metric qua Pushgateway; chỉ cập nhật thời điểm thành công khi có dữ liệu thật và dùng log có cấu trúc, không có PII.

Nguồn tham khảo

• When to use the Pushgateway, kiểm tra ngày 2026-10-28
• Metric types, kiểm tra ngày 2026-10-28