資料管線監控與異常偵測技術

– 10 分鐘閱讀

在資料工程領域，確保資料管線的穩定性和資料品質至關重要。本文將探討如何利用SQL查詢和視覺化工具來監控資料管線的健康狀況，並有效地偵測異常。我們將以一個模擬的系外行星資料集為例，逐步示範如何偵測NULL值比例異常、資料模式變更以及資料血統。首先，我們會利用SQL查詢計算特定欄位的NULL值比例，並透過設定閾值來觸發警示。接著，我們將探討如何使用移動平均值來更精確地偵測異常，避免單點異常造成的誤判。此外，我們還會示範如何追蹤資料模式的變更，例如新增或刪除欄位，以及如何利用資料血統視覺化工具來理解資料的流動和轉換過程。最後，我們將討論一些進階的異常偵測技術，例如結合機器學習演算法來提升偵測的準確性和效率。

建立異常偵測演算法

在資料管線的監控與異常偵測過程中，我們不僅僅需要識別出異常的指標，更需要結合時間序列中的重要上下文資訊，例如「當天是星期幾？」或是「這種模式是否會重複出現？」。並且，並非所有異常觀測值都是有趣的，它們不一定能幫助我們識別和糾正資料停擺的問題。

異常分佈的查詢

範例4-8展示了一條查詢異常分佈的SQL指令。

SELECT
    DATE_ADDED,
    CAST(
        SUM(
            CASE
                WHEN DISTANCE IS NULL THEN 1
                ELSE 0
            END
        ) AS FLOAT) / COUNT(*) AS DISTANCE_NULL_RATE,
    CAST(
        SUM(
            CASE
                WHEN G IS NULL THEN 1
                ELSE 0
        ) AS FLOAT) / COUNT(*) AS G_NULL_RATE,
    CAST(
        SUM(
            CASE
                WHEN ORBITAL_PERIOD IS NULL THEN 1
                ELSE 0
        ) AS FLOAT) / COUNT(*) AS ORBITAL_PERIOD_NULL_RATE,
    CAST(
        SUM(
            CASE
                WHEN AVG_TEMP IS NULL THEN 1
                ELSE 0
        ) AS FLOAT) / COUNT(*) AS AVG_TEMP_NULL_RATE
FROM
    EXOPLANETS
GROUP BY
    DATE_ADDED;

內容解密：

此查詢陳述式用於計算EXOPLANETS表中各個欄位（DISTANCE、G、ORBITAL_PERIOD、AVG_TEMP）的NULL值比例，並按照DATE_ADDED進行分組。這樣可以幫助我們瞭解每天新增資料中，各欄位的NULL值比例，從而進一步分析資料品質。

查詢結果分析

範例4-9展示了範例4-8查詢結果的部分資料。

date_addedDISTANCE_NULL_RATEG_NULL_RATEORBITAL_PERIOD_NULL_RATE
2020-01-010.08333333333333330.1785714285714290.214285714285714
2020-01-020.00.1521739130434780.326086956521739
…………

透過視覺化這些資料（Figure4-8），我們可以清楚地看到哪些日期的NULL值比例異常高。

簡單的異常偵測

範例4-10展示了一條簡單的SQL查詢，用於偵測AVG_TEMP欄位中的NULL值比例是否超過0.9。

WITH NULL_RATES AS (
    SELECT
        DATE_ADDED,
        CAST(
            SUM(
                CASE
                    WHEN AVG_TEMP IS NULL THEN 1
                    ELSE 0
                END
            ) AS FLOAT) / COUNT(*) AS AVG_TEMP_NULL_RATE
    FROM
        EXOPLANETS
    GROUP BY
        DATE_ADDED
)
SELECT
    *
FROM
    NULL_RATES
WHERE
    AVG_TEMP_NULL_RATE > 0.9;

內容解密：

此查詢首先計算每天AVG_TEMP欄位的NULL值比例，然後篩選出比例大於0.9的日期。這樣可以幫助我們快速識別出哪些天數的資料品質存在嚴重問題。

篩選連續異常日期

範例4-12展示了一條SQL查詢，用於篩選出連續多天出現異常NULL值比例的日期。

WITH NULL_RATES AS (
    SELECT
        DATE_ADDED,
        CAST(
            SUM(
                CASE
                    WHEN AVG_TEMP IS NULL THEN 1
                    ELSE 0
                END
            ) AS FLOAT) / COUNT(*) AS AVG_TEMP_NULL_RATE
    FROM
        EXOPLANETS
    GROUP BY
        DATE_ADDED
),
ALL_DATES AS (
    SELECT
        *,
        JULIANDAY(DATE_ADDED) - JULIANDAY(LAG(DATE_ADDED) OVER (ORDER BY DATE_ADDED)) AS DAYS_SINCE_LAST_ALERT
    FROM
        NULL_RATES
    WHERE
        AVG_TEMP_NULL_RATE > 0.9
)
SELECT
    DATE_ADDED,
    AVG_TEMP_NULL_RATE
FROM
    ALL_DATES
WHERE
    DAYS_SINCE_LAST_ALERT IS NULL OR DAYS_SINCE_LAST_ALERT > 1;

內容解密：

此查詢首先計算每天AVG_TEMP欄位的NULL值比例，然後篩選出比例大於0.9的日期。接著，透過計算日期之間的差值，篩選出不連續的異常日期。這樣可以幫助我們減少重複警示，提高異常偵測的準確性。

應用移動平均進行異常偵測

範例4-14展示了一條SQL查詢，用於應用移動平均來偵測異常NULL值比例。

WITH NULL_RATES AS (
    SELECT
        DATE_ADDED,
        CAST(
            SUM(
                CASE
                    WHEN AVG_TEMP IS NULL THEN 1
                    ELSE 0
                END
            ) AS FLOAT) / COUNT(*) AS AVG_TEMP_NULL_RATE
    FROM
        EXOPLANETS
    GROUP BY
        DATE_ADDED
),
NULL_WITH_AVG AS (
    SELECT
        *,
        AVG(AVG_TEMP_NULL_RATE) OVER (
            ORDER BY DATE_ADDED ASC
            ROWS BETWEEN 14 PRECEDING AND CURRENT ROW
        ) AS TWO_WEEK_ROLLING_AVG
    FROM
        NULL_RATES
)
SELECT
    *
FROM
    NULL_WITH_AVG
WHERE
    AVG_TEMP_NULL_RATE - TWO_WEEK_ROLLING_AVG > 0.3;

內容解密：

此查詢首先計算每天AVG_TEMP欄位的NULL值比例，然後計算過去兩週的移動平均值。最後，篩選出NULL值比例與移動平均值差值大於0.3的日期。這樣可以幫助我們更智慧地偵測異常NULL值比例，提高偵測的準確性。

建構資料管線的監控與異常偵測：探討綱要與沿襲監控

在前一節中，我們探討了資料可觀察性的前兩個支柱：資料的新鮮度（Freshness）與分佈（Distribution）。本文將進一步探索資料可觀察性的另兩個重要支柱：綱要（Schema）與沿襲（Lineage）。這兩個支柱對於理解資料異常的背景至關重要，能夠幫助我們識別異常發生的位置、上游資料管線的潛在影響，以及下游儀錶板的受影響程度。

針對綱要變更與沿襲的異常偵測

如同前一節，我們將繼續使用模擬的系外行星天文資料集。這個資料集最早的資料日期為2020-01-01，而最新的資料則是2020-07-18。為了進一步探索綱要與沿襲，我們需要擴充套件我們的資料環境。

sqlite> SELECT DATE_ADDED FROM EXOPLANETS ORDER BY DATE_ADDED DESC LIMIT 1;
2020-07-18

除了原有的EXOPLANETS表格，我們現在引入了一個新的表格EXOPLANETS_EXTENDED。這個表格是EXOPLANETS的超集，可以理解為同一個表格在不同時間點的狀態。EXOPLANETS_EXTENDED包含了從2020-01-01到2020-09-06的資料。

sqlite> SELECT DATE_ADDED FROM EXOPLANETS_EXTENDED ORDER BY DATE_ADDED ASC LIMIT 1;
2020-01-01

sqlite> SELECT DATE_ADDED FROM EXOPLANETS_EXTENDED ORDER BY DATE_ADDED DESC LIMIT 1;
2020-09-06

內容解密：

上述SQL查詢展示瞭如何檢查EXOPLANETS_EXTENDED表格中最早和最新的資料日期。這有助於我們理解資料的時間範圍。

新增欄位與綱要變更

EXOPLANETS_EXTENDED表格相較於EXOPLANETS，多了兩個欄位：ECCENTRICITY（軌道離心率）和ATMOSPHERE（大氣成分）。這兩個新增欄位為資料異常偵測提供了新的機會。

sqlite> PRAGMA TABLE_INFO(EXOPLANETS_EXTENDED);

輸出結果顯示了EXOPLANETS_EXTENDED表格的欄位資訊，包括欄位名稱、資料型別等。

內容解密：

  • _ID欄位是主鍵，資料型別為VARCHAR(16777216)。
  • DISTANCE、G、ORBITAL_PERIOD、AVG_TEMP欄位儲存了與行星相關的物理資料。
  • DATE_ADDED欄位記錄了資料新增的時間戳記。
  • ECCENTRICITY和ATMOSPHERE是新增的欄位，分別記錄了行星的軌道離心率和大氣成分。

檢查資料是否回填

由於資料並未回填（backfill），因此早期資料的ECCENTRICITY和ATMOSPHERE欄位可能為空。

SELECT 
    DATE_ADDED, 
    ECCENTRICITY, 
    ATMOSPHERE 
FROM 
    EXOPLANETS_EXTENDED 
ORDER BY 
    DATE_ADDED ASC 
LIMIT 10;

執行結果顯示了最早的十筆資料的DATE_ADDED、ECCENTRICITY和ATMOSPHERE欄位的值。可以觀察到早期資料的ECCENTRICITY和ATMOSPHERE欄位大多為空值。

內容解密：

這個查詢幫助我們瞭解資料的回填情況。由於資料未回填，早期資料的ECCENTRICITY和ATMOSPHERE欄位為空，這是綱要變更的結果。

資料表結構變化

  graph LR
    A[EXOPLANETS] -->|新增欄位|> B[EXOPLANETS_EXTENDED]
    B --> C{ECCENTRICITY 和 ATMOSPHERE 欄位}
    C -->|早期資料可能為空|> D[未回填資料]

圖表翻譯： 此圖表展示了EXOPLANETS表格擴充套件為EXOPLANETS_EXTENDED的過程，以及新增的ECCENTRICITY和ATMOSPHERE欄位。由於資料未回填，早期資料的這兩個欄位可能為空值。

未來方向

未來，我們可以進一步探索如何結合機器學習技術來增強資料異常偵測的能力。同時，也可以研究如何將這些技術應用於更大規模的資料集和更複雜的資料管線中。

透過持續改進和擴充套件資料可觀察性的技術，我們可以更好地應對日益增長的資料量和複雜性，確保資料的準確性和可靠性。

資料模式變更的監控與異常檢測

在資料處理和分析的過程中，資料模式（Schema）的變更是常見且可能引起問題的事件。資料模式變更指的是資料結構的改變，例如新增或刪除欄位、表格等。這類別變更可能導致資料處理流程中的異常，因此監控和檢測資料模式變更至關重要。

為什麼資料模式變更需要監控？

資料模式變更可能代表著資料來源的結構發生了變化，這種變化可能由多種原因引起，包括但不限於：

  • 新的API端點的加入
  • 預期被棄用但尚未被棄用的欄位
  • 資料表格中欄位、列或整個表格的新增或刪除

這些變更可能對資料處理流程產生重大影響，因此需要被及時發現和處理。

如何監控資料模式變更？

在理想情況下，我們希望能夠記錄資料模式變更的歷史。然而，許多資料函式庫預設並不具備版本控制功能，因此需要額外的機制來追蹤這些變更。

使用額外表格記錄資料模式變更

一種常見的做法是建立一個額外的表格來記錄資料模式的變更。例如，在前述的例子中，建立了一個名為EXOPLANETS_COLUMNS的表格來記錄EXOPLANETS_EXTENDED表格中欄位的變更。

透過查詢這個表格，可以發現資料模式變更的日期。具體方法是使用SQL查詢來比較不同日期下記錄的欄位資訊，從而找出變更發生的時間點。

WITH CHANGES AS (
  SELECT
    DATE,
    COLUMNS AS NEW_COLUMNS,
    LAG(COLUMNS) OVER (ORDER BY DATE) AS PAST_COLUMNS
  FROM
    EXOPLANETS_COLUMNS
)
SELECT
  *
FROM
  CHANGES
WHERE
  NEW_COLUMNS != PAST_COLUMNS
ORDER BY
  DATE ASC;

查詢結果分析

上述查詢傳回的結果顯示了資料模式變更的日期和變更前後的欄位資訊。透過分析這些資訊，可以確定資料模式變更的具體內容和時間。

DATE: 2020-07-19
NEW_COLUMNS: [
  (0, '_id', 'TEXT', 0, None, 0),
  (1, 'distance', 'REAL', 0, None, 0),
  (2, 'g', 'REAL', 0, None, 0),
  (3, 'orbital_period', 'REAL', 0, None, 0),
  (4, 'avg_temp', 'REAL', 0, None, 0),
  (5, 'date_added', 'TEXT', 0, None, 0),
  (6, 'eccentricity', 'REAL', 0, None, 0),
  (7, 'atmosphere', 'TEXT', 0, None, 0)
]
PAST_COLUMNS: [
  (0, '_id', 'TEXT', 0, None, 0),
  (1, 'distance', 'REAL', 0, None, 0),
  (2, 'g', 'REAL', 0, None, 0),
  (3, 'orbital_period', 'REAL', 0, None, 0),
  (4, 'avg_temp', 'REAL', 0, None, 0),
  (5, 'date_added', 'TEXT', 0, None, 0)
]

資料模式變更檢測的通用方法

實作資料模式變更的監控和檢測遵循一個通用模式：識別有用的中繼資料（Metadata），追蹤這些中繼資料，並建立檢測器來提醒潛在的問題。除了使用額外的表格來記錄資料模式變更外，還有多種其他方法可以實作資料模式變更的檢測。

資料血統（Lineage）的視覺化

資料血統是資料可觀察性的五大支柱中最全面的概念。它透過描述資料的來源、流動和轉換過程，幫助我們理解資料的背景和關聯。視覺化資料血統可以幫助我們更好地理解資料的流動和依賴關係，從而更好地監控和管理資料處理流程。

使用SQL查詢展示資料血統

雖然使用SQL查詢來視覺化資料血統並不直觀，但透過簡單的例子可以展示其價值。假設我們有一個名為HABITABLES的表格，它根據EXOPLANETS表格中的資料計算出行星是否適合居住。

sqlite> PRAGMA TABLE_INFO(HABITABLES);
_id | TEXT | 0 | | 0
perihelion | REAL | 0 | | 0
aphelion | REAL | 0 | | 0
atmosphere | TEXT | 0 | | 0
habitability | REAL | 0 | | 0
min_temp | REAL | 0 | | 0

透過分析HABITABLES和EXOPLANETS表格之間的關聯，可以瞭解資料的流動和轉換過程。

未來，我們可以進一步探索和開發更多的方法和工具來監控和檢測資料模式變更。同時，透過視覺化資料血統，可以更好地理解和最佳化資料處理流程，提高資料的可觀察性和可管理性。

實際應用案例

在實際應用中，我們可以根據具體的業務需求和資料特點，選擇合適的方法和工具來監控和檢測資料模式變更。例如，在資料倉函式庫建設中，可以使用資料血統視覺化工具來展示資料的流動和轉換過程，從而更好地理解和管理資料處理流程。

技術選型考量

在選擇資料模式變更監控和檢測工具時，需要考慮多種因素，包括資料函式庫型別、資料量、處理效能等。同時，也需要考慮工具的可擴充套件性和可維護性，以確保能夠滿足未來業務發展的需求。

程式碼範例

import sqlite3

# 連線資料函式庫
conn = sqlite3.connect('exoplanets.db')
cursor = conn.cursor()

# 查詢資料模式變更
def query_schema_change():
    query = """
        WITH CHANGES AS (
          SELECT
            DATE,
            COLUMNS AS NEW_COLUMNS,
            LAG(COLUMNS) OVER (ORDER BY DATE) AS PAST_COLUMNS
          FROM
            EXOPLANETS_COLUMNS
        )
        SELECT
          *
        FROM
          CHANGES
        WHERE
          NEW_COLUMNS != PAST_COLUMNS
        ORDER BY
          DATE ASC;
    """
    cursor.execute(query)
    results = cursor.fetchall()
    return results

# 列印查詢結果
results = query_schema_change()
for row in results:
    print(row)

# 關閉資料函式庫連線
conn.close()

內容解密：

此Python程式碼範例展示瞭如何連線SQLite資料函式庫並查詢資料模式變更。程式碼首先匯入必要的sqlite3模組，然後建立與資料函式庫的連線。query_schema_change函式定義了一個SQL查詢，用於檢測資料模式變更。查詢結果透過fetchall方法取得並傳回。最後，程式碼列印查詢結果並關閉資料函式庫連線。

圖表翻譯

此圖表展示了資料模式變更的檢測過程。

  graph LR;
    A[開始] --> B[連線資料函式庫];
    B --> C[執行查詢];
    C --> D[取得查詢結果];
    D --> E[分析結果];
    E --> F[輸出結果];

圖表翻譯： 此圖表描述了資料模式變更檢測的流程。首先，系統連線資料函式庫，然後執行SQL查詢以檢測變更。查詢結果被取得並分析，最後輸出結果。這個過程展示瞭如何系統化地監控資料模式變更。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。