監控資料管線

使用 Python 的 ETL 與 ELT

Jake Roach

Data Engineer

監控資料管線

應監控資料管線的資料變動與執行失敗

  • 遺漏資料
  • 資料型別漂移
  • 套件棄用或功能變更

$$

資料管線的記錄流程。

使用 Python 的 ETL 與 ELT

記錄資料管線效能

  • 在執行時記錄效能
  • 失敗時提供排查起點
import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.debug(f"Variable has value {path}")
logging.info("Data has been transformed and will now be loaded.")
DEBUG: Variable has value raw_file.csv
INFO: Data has been transformed and will now be loaded.
使用 Python 的 ETL 與 ELT

記錄警告與錯誤

import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.warning("Unexpected number of rows detected.")
logging.error("{ke} arose in execution.")
WARNING: Unexpected number of rows detected.
ERROR: KeyError arose in execution.
使用 Python 的 ETL 與 ELT

用 try-except 處理例外

try:
    # Execute some code here
    ...

except:
    # Logging about failures that occured
    # Logic to execute upon exception
    ...

  • 發生錯誤時可執行對應程式碼
使用 Python 的 ETL 與 ELT

用 try-except 處理特定例外

except 子句中指定例外類型

try:
    # Try to filter by price_change
    clean_stock_data = transform(raw_stock_data)
    logging.info("Successfully filtered DataFrame by 'price_change'")

except KeyError as ke:
    # Handle the error, create new column, transform
    logging.warning(f"{ke}: Cannot filter DataFrame by 'price_change'")
    raw_stock_data["price_change"] = raw_stock_data["close"] - raw_stock_data["open"]
    clean_stock_data = transform(raw_stock_data)

使用 Python 的 ETL 與 ELT

一起來練習吧!

使用 Python 的 ETL 與 ELT

Preparing Video For Download...