Memantau pipeline data

ETL dan ELT di Python

Jake Roach

Data Engineer

Memantau pipeline data

Pipeline data harus dipantau untuk perubahan data dan kegagalan eksekusi

Data hilang
Pergeseran tipe data
Depresiasi paket atau perubahan fungsionalitas

Alur pencatatan log untuk sebuah pipeline data.

Mencatat kinerja pipeline data

Dokumentasikan kinerja saat eksekusi
Menjadi titik awal ketika solusi gagal

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.

Mencatat peringatan dan error

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.

Menangani pengecualian dengan try-except

try:
    # Execute some code here
    ...

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

Menyediakan cara mengeksekusi kode jika terjadi error

Menangani pengecualian spesifik dengan try-except

Masukkan pengecualian spesifik pada klausa 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)

Ayo berlatih!

ETL dan ELT di Python